7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include "lib.h"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include "array.h"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include "hex-binary.h"

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen#include "str.h"

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen#include "ioloop.h"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include "net.h"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include "write-full.h"

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen#include "time-util.h"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include "var-expand.h"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include "settings-parser.h"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include "sql-api-private.h"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen#ifdef BUILD_CASSANDRA

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include <fcntl.h>

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#include <unistd.h>

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen#include <cassandra.h>

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#define IS_CONNECTED(db) \

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ((db)->api.state != SQL_DB_STATE_DISCONNECTED && \

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen (db)->api.state != SQL_DB_STATE_CONNECTING)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#define CASSANDRA_FALLBACK_WARN_INTERVAL_SECS 60

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#define CASSANDRA_FALLBACK_FIRST_RETRY_MSECS 50

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#define CASSANDRA_FALLBACK_MAX_RETRY_MSECS (1000*60)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#define CASS_QUERY_DEFAULT_WARN_TIMEOUT_MSECS (5*1000)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainentypedef void driver_cassandra_callback_t(CassFuture *future, void *context);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenenum cassandra_counter_type {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_SENT,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_RECV_OK,

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_NO_HOSTS,

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_QUEUE_FULL,

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_CLIENT_TIMEOUT,

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_SERVER_TIMEOUT,

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_SERVER_UNAVAILABLE,

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_OTHER,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CASSANDRA_COUNTER_TYPE_QUERY_SLOW,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CASSANDRA_COUNTER_COUNT

0c47c2096714b50880d48d00ce0bf28349eb4aceTimo Sirainen};

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic const char *counter_names[CASSANDRA_COUNTER_COUNT] = {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen "sent",

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen "recv_ok",

9eef11df882f9c14d164f42cb438f32fe724041cTimo Sirainen "recv_err_no_hosts",

9eef11df882f9c14d164f42cb438f32fe724041cTimo Sirainen "recv_err_queue_full",

9eef11df882f9c14d164f42cb438f32fe724041cTimo Sirainen "recv_err_client_timeout",

9eef11df882f9c14d164f42cb438f32fe724041cTimo Sirainen "recv_err_server_timeout",

678d0463849ba777106eb7875f27db07a5d8e3dfTimo Sirainen "recv_err_server_unavailable",

a75d470c9223a75801418fcdda258885c36317e0Timo Sirainen "recv_err_other",

678d0463849ba777106eb7875f27db07a5d8e3dfTimo Sirainen "slow",

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen};

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenenum cassandra_query_type {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CASSANDRA_QUERY_TYPE_READ,

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen CASSANDRA_QUERY_TYPE_READ_MORE,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CASSANDRA_QUERY_TYPE_WRITE,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CASSANDRA_QUERY_TYPE_DELETE,

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CASSANDRA_QUERY_TYPE_COUNT

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen};

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic const char *cassandra_query_type_names[CASSANDRA_QUERY_TYPE_COUNT] = {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen "read", "read-more", "write", "delete"

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen};

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainenstruct cassandra_callback {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int id;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CassFuture *future;

678d0463849ba777106eb7875f27db07a5d8e3dfTimo Sirainen struct cassandra_db *db;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen driver_cassandra_callback_t *callback;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen void *context;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen};

4ee00532a265bdfb38539d811fcd12d51210ac35Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstruct cassandra_db {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct sql_db api;

678d0463849ba777106eb7875f27db07a5d8e3dfTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen char *hosts, *keyspace, *user, *password;

678d0463849ba777106eb7875f27db07a5d8e3dfTimo Sirainen CassConsistency read_consistency, write_consistency, delete_consistency;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CassConsistency read_fallback_consistency, write_fallback_consistency, delete_fallback_consistency;

4ee00532a265bdfb38539d811fcd12d51210ac35Timo Sirainen CassLogLevel log_level;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen bool debug_queries;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen bool latency_aware_routing;

8b3a4836da0b032673918941cb49c956d3b89b25Timo Sirainen unsigned int protocol_version;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int num_threads;

4ee00532a265bdfb38539d811fcd12d51210ac35Timo Sirainen unsigned int connect_timeout_msecs, request_timeout_msecs;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int warn_timeout_msecs;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int heartbeat_interval_secs, idle_timeout_secs;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int execution_retry_interval_msecs, execution_retry_times;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen unsigned int page_size;

47255691575e06a1c95ce78ff0a1b502199de3abTimo Sirainen in_port_t port;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CassCluster *cluster;

307ec6c2c319e3335ddb1a7aca2d2884fe17fae0Timo Sirainen CassSession *session;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen CassTimestampGen *timestamp_gen;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen int fd_pipe[2];

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen struct io *io_pipe;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ARRAY(struct cassandra_callback *) callbacks;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ARRAY(struct cassandra_result *) results;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int callback_ids;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen char *metrics_path;

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen struct timeout *to_metrics;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen uint64_t counters[CASSANDRA_COUNTER_COUNT];

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen struct timeval primary_query_last_sent[CASSANDRA_QUERY_TYPE_COUNT];

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen time_t last_fallback_warning[CASSANDRA_QUERY_TYPE_COUNT];

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen unsigned int fallback_failures[CASSANDRA_QUERY_TYPE_COUNT];

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen /* for synchronous queries: */

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen struct ioloop *ioloop, *orig_ioloop;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen struct sql_result *sync_result;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen char *error;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen};

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainenstruct cassandra_result {

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen struct sql_result api;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen CassStatement *statement;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen const CassResult *result;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen CassIterator *iterator;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen char *query;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen char *error;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen CassConsistency consistency, fallback_consistency;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen enum cassandra_query_type query_type;

31d32d39dd09be0625a6d92ee715155f5d679515Timo Sirainen struct timeval page0_start_time, start_time, finish_time;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int row_count, total_row_count, page_num;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen pool_t row_pool;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ARRAY_TYPE(const_string) fields;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ARRAY(size_t) field_sizes;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen sql_query_callback_t *callback;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen void *context;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen bool query_sent:1;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen bool finished:1;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen bool paging_continues:1;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen};

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstruct cassandra_transaction_context {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct sql_transaction_context ctx;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen int refcount;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen sql_commit_callback_t *callback;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen void *context;

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen pool_t query_pool;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen char *error;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen bool begin_succeeded:1;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen bool begin_failed:1;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen bool failed:1;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen};

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainenextern const struct sql_db driver_cassandra_db;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainenextern const struct sql_result driver_cassandra_result;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainenstatic struct {

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen CassConsistency consistency;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen const char *name;

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen} cass_consistency_names[] = {

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen { CASS_CONSISTENCY_ANY, "any" },

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen { CASS_CONSISTENCY_ONE, "one" },

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen { CASS_CONSISTENCY_TWO, "two" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_CONSISTENCY_THREE, "three" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_CONSISTENCY_QUORUM, "" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_CONSISTENCY_ALL, "all" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_CONSISTENCY_LOCAL_QUORUM, "local-quorum" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_CONSISTENCY_EACH_QUORUM, "each-quorum" },

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen { CASS_CONSISTENCY_SERIAL, "serial" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_CONSISTENCY_LOCAL_SERIAL, "local-serial" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_CONSISTENCY_LOCAL_ONE, "local-one" }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen};

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic struct {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CassLogLevel log_level;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen const char *name;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen} cass_log_level_names[] = {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_LOG_CRITICAL, "critical" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_LOG_ERROR, "error" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_LOG_WARN, "warn" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_LOG_INFO, "info" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_LOG_DEBUG, "debug" },

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { CASS_LOG_TRACE, "trace" }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen};

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void driver_cassandra_result_send_query(struct cassandra_result *result);

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainenstatic void driver_cassandra_send_queries(struct cassandra_db *db);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void result_finish(struct cassandra_result *result);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic int consistency_parse(const char *str, CassConsistency *consistency_r)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int i;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen for (i = 0; i < N_ELEMENTS(cass_consistency_names); i++) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (strcmp(cass_consistency_names[i].name, str) == 0) {

678d0463849ba777106eb7875f27db07a5d8e3dfTimo Sirainen *consistency_r = cass_consistency_names[i].consistency;

678d0463849ba777106eb7875f27db07a5d8e3dfTimo Sirainen return 0;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return -1;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic int log_level_parse(const char *str, CassLogLevel *log_level_r)

ec0cc8fa647794e44a1afaa448f495a713048dc4Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int i;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen for (i = 0; i < N_ELEMENTS(cass_log_level_names); i++) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (strcmp(cass_log_level_names[i].name, str) == 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen *log_level_r = cass_log_level_names[i].log_level;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return 0;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen }

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen return -1;

307ec6c2c319e3335ddb1a7aca2d2884fe17fae0Timo Sirainen}

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainenstatic void driver_cassandra_set_state(struct cassandra_db *db, enum sql_db_state state)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen /* switch back to original ioloop in case the caller wants to

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->ioloop != NULL)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen io_loop_set_current(db->orig_ioloop);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen sql_db_set_state(&db->api, state);

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen if (db->ioloop != NULL)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen io_loop_set_current(db->ioloop);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

08e9fec5ba9e1a26e658c4224207d666b6ced27dTimo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainenstatic void driver_cassandra_close(struct cassandra_db *db, const char *error)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_result *const *resultp;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen if (db->io_pipe != NULL)

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen io_remove(&db->io_pipe);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen if (db->fd_pipe[0] != -1) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_close_fd(&db->fd_pipe[0]);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen i_close_fd(&db->fd_pipe[1]);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen }

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen driver_cassandra_set_state(db, SQL_DB_STATE_DISCONNECTED);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen while (array_count(&db->results) > 0) {

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen resultp = array_idx(&db->results, 0);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen if ((*resultp)->error == NULL)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen (*resultp)->error = i_strdup(error);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen result_finish(*resultp);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen }

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen if (db->ioloop != NULL) {

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen /* running a sync query, stop it */

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen io_loop_stop(db->ioloop);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen }

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen}

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainenstatic void driver_cassandra_log_error(CassFuture *future, const char *str)

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen{

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen const char *message;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen size_t size;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen cass_future_error_message(future, &message, &size);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen i_error("cassandra: %s: %.*s", str, (int)size, message);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen}

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainenstatic void driver_cassandra_future_callback(CassFuture *future ATTR_UNUSED,

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen void *context)

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen{

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen struct cassandra_callback *cb = context;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen /* this isn't the main thread - communicate with main thread by

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen if (write_full(cb->db->fd_pipe[1], &cb->id, sizeof(cb->id)) < 0) {

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen const char *str = t_strdup_printf(

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen "cassandra: write(pipe) failed: %s\n",

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen strerror(errno));

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen (void)write_full(STDERR_FILENO, str, strlen(str));

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen }

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen}

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainenstatic void cassandra_callback_run(struct cassandra_callback *cb)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen{

5e32f3d9e2c58b6db53cc3b063c9ee73949207caTimo Sirainen cb->callback(cb->future, cb->context);

5e32f3d9e2c58b6db53cc3b063c9ee73949207caTimo Sirainen cass_future_free(cb->future);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_free(cb);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen}

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainenstatic void driver_cassandra_input_id(struct cassandra_db *db, unsigned int id)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen{

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen struct cassandra_callback *cb, *const *cbp;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen /* usually there are only a few callbacks, so don't bother with using

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen array_foreach(&db->callbacks, cbp) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cb = *cbp;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen if (cb->id == id) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen array_delete(&db->callbacks,

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen array_foreach_idx(&db->callbacks, cbp), 1);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cassandra_callback_run(cb);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen return;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen }

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen }

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_panic("cassandra: Received unknown ID %u", id);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen}

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainenstatic void driver_cassandra_input(struct cassandra_db *db)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen{

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen unsigned int ids[1024];

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen ssize_t ret;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen ret = read(db->fd_pipe[0], ids, sizeof(ids));

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen if (ret < 0)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_error("cassandra: read(pipe) failed: %m");

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen else if (ret == 0)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_error("cassandra: read(pipe) failed: EOF");

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen else if (ret % sizeof(ids[0]) != 0)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_error("cassandra: read(pipe) returned wrong amount of data");

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen else {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen /* success */

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen unsigned int i, count = ret / sizeof(ids[0]);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen for (i = 0; i < count && db->api.state != SQL_DB_STATE_DISCONNECTED; i++)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen driver_cassandra_input_id(db, ids[i]);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen return;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen }

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen driver_cassandra_close(db, "IPC pipe closed");

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen}

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainenstatic void

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainendriver_cassandra_set_callback(CassFuture *future, struct cassandra_db *db,

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen driver_cassandra_callback_t *callback,

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen void *context)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen{

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen struct cassandra_callback *cb;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cb = i_new(struct cassandra_callback, 1);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cb->id = ++db->callback_ids;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cb->future = future;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cb->callback = callback;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cb->context = context;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cb->db = db;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen array_append(&db->callbacks, &cb, 1);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen cass_future_set_callback(future, driver_cassandra_future_callback, cb);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen}

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainenstatic void connect_callback(CassFuture *future, void *context)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen{

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen struct cassandra_db *db = context;

ab3c52cff40218f248fac2bd7c93125cc2ae4c9dTimo Sirainen CassError rc;

ab3c52cff40218f248fac2bd7c93125cc2ae4c9dTimo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen if ((rc = cass_future_error_code(future)) != CASS_OK) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen driver_cassandra_log_error(future,

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen "Couldn't connect to Cassandra");

ab3c52cff40218f248fac2bd7c93125cc2ae4c9dTimo Sirainen driver_cassandra_close(db, "Couldn't connect to Cassandra");

ab3c52cff40218f248fac2bd7c93125cc2ae4c9dTimo Sirainen return;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen }

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen driver_cassandra_set_state(db, SQL_DB_STATE_IDLE);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen if (db->ioloop != NULL) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen /* driver_cassandra_sync_init() waiting for connection to

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen io_loop_stop(db->ioloop);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen }

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen driver_cassandra_send_queries(db);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen}

9307e4f91ea313a303b5473d9c53a3a2f383300fTimo Sirainen

719123a3ec5aeb45ef1c50c265039666c71981d7Timo Sirainenstatic int driver_cassandra_connect(struct sql_db *_db)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen{

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen CassFuture *future;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen i_assert(db->api.state == SQL_DB_STATE_DISCONNECTED);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen if (pipe(db->fd_pipe) < 0) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_error("pipe() failed: %m");

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen return -1;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen }

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen db->io_pipe = io_add(db->fd_pipe[0], IO_READ,

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen driver_cassandra_input, db);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen driver_cassandra_set_state(db, SQL_DB_STATE_CONNECTING);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen future = cass_session_connect_keyspace(db->session, db->cluster, db->keyspace);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen driver_cassandra_set_callback(future, db, connect_callback, db);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen return 0;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen}

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainenstatic void driver_cassandra_disconnect(struct sql_db *_db)

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen{

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen driver_cassandra_close(db, "Disconnected");

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen}

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainenstatic const char *

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainendriver_cassandra_escape_string(struct sql_db *db ATTR_UNUSED,

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen const char *string)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen{

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen string_t *escaped;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen unsigned int i;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen if (strchr(string, '\'') == NULL)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen return string;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen escaped = t_str_new(strlen(string)+10);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen for (i = 0; string[i] != '\0'; i++) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen if (string[i] == '\'')

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen str_append_c(escaped, '\'');

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen str_append_c(escaped, string[i]);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen }

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen return str_c(escaped);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen}

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainenstatic void driver_cassandra_parse_connect_string(struct cassandra_db *db,

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen const char *connect_string)

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen{

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen const char *const *args, *key, *value, *error;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen string_t *hosts = t_str_new(64);

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen bool read_fallback_set = FALSE, write_fallback_set = FALSE, delete_fallback_set = FALSE;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen db->log_level = CASS_LOG_WARN;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen db->read_consistency = CASS_CONSISTENCY_LOCAL_QUORUM;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen db->write_consistency = CASS_CONSISTENCY_LOCAL_QUORUM;

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen db->delete_consistency = CASS_CONSISTENCY_LOCAL_QUORUM;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen db->connect_timeout_msecs = SQL_CONNECT_TIMEOUT_SECS*1000;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen db->request_timeout_msecs = SQL_QUERY_TIMEOUT_SECS*1000;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen db->warn_timeout_msecs = CASS_QUERY_DEFAULT_WARN_TIMEOUT_MSECS;

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen args = t_strsplit_spaces(connect_string, " ");

9307e4f91ea313a303b5473d9c53a3a2f383300fTimo Sirainen for (; *args != NULL; args++) {

9307e4f91ea313a303b5473d9c53a3a2f383300fTimo Sirainen value = strchr(*args, '=');

9307e4f91ea313a303b5473d9c53a3a2f383300fTimo Sirainen if (value == NULL) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_fatal("cassandra: Missing value in connect string: %s",

9307e4f91ea313a303b5473d9c53a3a2f383300fTimo Sirainen *args);

719123a3ec5aeb45ef1c50c265039666c71981d7Timo Sirainen }

719123a3ec5aeb45ef1c50c265039666c71981d7Timo Sirainen key = t_strdup_until(*args, value++);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen if (strcmp(key, "host") == 0) {

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen if (str_len(hosts) > 0)

719123a3ec5aeb45ef1c50c265039666c71981d7Timo Sirainen str_append_c(hosts, ',');

719123a3ec5aeb45ef1c50c265039666c71981d7Timo Sirainen str_append(hosts, value);

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen } else if (strcmp(key, "port") == 0) {

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen if (net_str2port(value, &db->port) < 0)

41e51b972f02e8b16c19fab9160294ea0a07c343Timo Sirainen i_fatal("cassandra: Invalid port: %s", value);

719123a3ec5aeb45ef1c50c265039666c71981d7Timo Sirainen } else if (strcmp(key, "dbname") == 0 ||

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen strcmp(key, "keyspace") == 0) {

55d33f807765482eb47374aaaced1fe714e0b256Timo Sirainen i_free(db->keyspace);

e34d170f8f0e084bd94bfbc1a7085ece67e508dfTimo Sirainen db->keyspace = i_strdup(value);

e34d170f8f0e084bd94bfbc1a7085ece67e508dfTimo Sirainen } else if (strcmp(key, "user") == 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_free(db->user);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->user = i_strdup(value);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "password") == 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_free(db->password);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->password = i_strdup(value);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "read_consistency") == 0) {

e34d170f8f0e084bd94bfbc1a7085ece67e508dfTimo Sirainen if (consistency_parse(value, &db->read_consistency) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Unknown read_consistency: %s", value);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen } else if (strcmp(key, "read_fallback_consistency") == 0) {

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen if (consistency_parse(value, &db->read_fallback_consistency) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Unknown read_fallback_consistency: %s", value);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen read_fallback_set = TRUE;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "write_consistency") == 0) {

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen if (consistency_parse(value, &db->write_consistency) < 0)

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen i_fatal("cassandra: Unknown write_consistency: %s", value);

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen } else if (strcmp(key, "write_fallback_consistency") == 0) {

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen if (consistency_parse(value, &db->write_fallback_consistency) < 0)

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen i_fatal("cassandra: Unknown write_fallback_consistency: %s", value);

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen write_fallback_set = TRUE;

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen } else if (strcmp(key, "delete_consistency") == 0) {

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen if (consistency_parse(value, &db->delete_consistency) < 0)

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen i_fatal("cassandra: Unknown delete_consistency: %s", value);

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen } else if (strcmp(key, "delete_fallback_consistency") == 0) {

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen if (consistency_parse(value, &db->delete_fallback_consistency) < 0)

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen i_fatal("cassandra: Unknown delete_fallback_consistency: %s", value);

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen delete_fallback_set = TRUE;

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen } else if (strcmp(key, "log_level") == 0) {

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen if (log_level_parse(value, &db->log_level) < 0)

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen i_fatal("cassandra: Unknown log_level: %s", value);

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen } else if (strcmp(key, "debug_queries") == 0) {

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen db->debug_queries = TRUE;

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen } else if (strcmp(key, "latency_aware_routing") == 0) {

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen db->latency_aware_routing = TRUE;

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen } else if (strcmp(key, "version") == 0) {

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen if (str_to_uint(value, &db->protocol_version) < 0)

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen i_fatal("cassandra: Invalid version: %s", value);

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen } else if (strcmp(key, "num_threads") == 0) {

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen if (str_to_uint(value, &db->num_threads) < 0)

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen i_fatal("cassandra: Invalid num_threads: %s", value);

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen } else if (strcmp(key, "heartbeat_interval") == 0) {

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen if (settings_get_time(value, &db->heartbeat_interval_secs, &error) < 0)

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen i_fatal("cassandra: Invalid heartbeat_interval '%s': %s", value, error);

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen } else if (strcmp(key, "idle_timeout") == 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (settings_get_time(value, &db->idle_timeout_secs, &error) < 0)

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen i_fatal("cassandra: Invalid idle_timeout '%s': %s", value, error);

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen } else if (strcmp(key, "connect_timeout") == 0) {

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen if (settings_get_time_msecs(value, &db->connect_timeout_msecs, &error) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Invalid connect_timeout '%s': %s", value, error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "request_timeout") == 0) {

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen if (settings_get_time_msecs(value, &db->request_timeout_msecs, &error) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Invalid request_timeout '%s': %s", value, error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "warn_timeout") == 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (settings_get_time_msecs(value, &db->warn_timeout_msecs, &error) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Invalid warn_timeout '%s': %s", value, error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "metrics") == 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_free(db->metrics_path);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->metrics_path = i_strdup(value);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "execution_retry_interval") == 0) {

86bde2c1838d1ce967fa2b394bb952004a4adcb7Timo Sirainen if (settings_get_time_msecs(value, &db->execution_retry_interval_msecs, &error) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Invalid execution_retry_interval '%s': %s", value, error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#ifndef HAVE_CASSANDRA_SPECULATIVE_POLICY

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: This cassandra version does not support execution_retry_interval");

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen#endif

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "execution_retry_times") == 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (str_to_uint(value, &db->execution_retry_times) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Invalid execution_retry_times %s", value);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#ifndef HAVE_CASSANDRA_SPECULATIVE_POLICY

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen i_fatal("cassandra: This cassandra version does not support execution_retry_times");

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen#endif

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (strcmp(key, "page_size") == 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (str_to_uint(value, &db->page_size) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Invalid page_size: %s", value);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("cassandra: Unknown connect string: %s", key);

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen }

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen }

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen if (!read_fallback_set)

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen db->read_fallback_consistency = db->read_consistency;

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen if (!write_fallback_set)

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen db->write_fallback_consistency = db->write_consistency;

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen if (!delete_fallback_set)

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen db->delete_fallback_consistency = db->delete_consistency;

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen if (str_len(hosts) == 0)

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen i_fatal("cassandra: No hosts given in connect string");

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen if (db->keyspace == NULL)

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen i_fatal("cassandra: No dbname given in connect string");

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen db->hosts = i_strdup(str_c(hosts));

a1893588a73d32a322153f5a2f7a16ef2c724979Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainendriver_cassandra_get_metrics_json(struct cassandra_db *db, string_t *dest)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen#define ADD_UINT64(_struct, _field) \

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen str_printfa(dest, "\""#_field"\": %llu,", (unsigned long long)metrics._struct._field);

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen#define ADD_DOUBLE(_struct, _field) \

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen str_printfa(dest, "\""#_field"\": %02lf,", metrics._struct._field);

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen CassMetrics metrics;

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen cass_session_get_metrics(db->session, &metrics);

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen str_append(dest, "{ \"requests\": {");

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen ADD_UINT64(requests, min);

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen ADD_UINT64(requests, max);

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen ADD_UINT64(requests, mean);

d52f5dcb05092e126058874772f2c367499e650aTimo Sirainen ADD_UINT64(requests, stddev);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(requests, median);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(requests, percentile_75th);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(requests, percentile_95th);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(requests, percentile_98th);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(requests, percentile_99th);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(requests, percentile_999th);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_DOUBLE(requests, mean_rate);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_DOUBLE(requests, one_minute_rate);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_DOUBLE(requests, five_minute_rate);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_DOUBLE(requests, fifteen_minute_rate);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_truncate(dest, str_len(dest)-1);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_append(dest, "}, \"stats\": {");

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(stats, total_connections);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(stats, available_connections);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(stats, exceeded_pending_requests_water_mark);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(stats, exceeded_write_bytes_water_mark);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_truncate(dest, str_len(dest)-1);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_append(dest, "}, \"errors\": {");

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(errors, connection_timeouts);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(errors, pending_request_timeouts);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen ADD_UINT64(errors, request_timeouts);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_truncate(dest, str_len(dest)-1);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen str_append(dest, "}, \"queries\": {");

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen for (unsigned int i = 0; i < CASSANDRA_COUNTER_COUNT; i++) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_printfa(dest, "\"%s\": %llu,", counter_names[i],

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen (unsigned long long)db->counters[i]);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_truncate(dest, str_len(dest)-1);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_append(dest, "}}");

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void driver_cassandra_metrics_write(struct cassandra_db *db)

678d0463849ba777106eb7875f27db07a5d8e3dfTimo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct var_expand_table tab[] = {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen { '\0', NULL, NULL }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen };

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen string_t *path = t_str_new(64);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen string_t *data;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen const char *error;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen int fd;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen if (var_expand(path, db->metrics_path, tab, &error) <= 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_error("cassandra: Failed to expand metrics_path=%s: %s",

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen db->metrics_path, error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen fd = open(str_c(path), O_WRONLY | O_CREAT | O_TRUNC | O_NONBLOCK, 0600);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (fd == -1) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_error("creat(%s) failed: %m", str_c(path));

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen data = t_str_new(1024);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen driver_cassandra_get_metrics_json(db, data);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (write_full(fd, str_data(data), str_len(data)) < 0)

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen i_error("write(%s) failed: %m", str_c(path));

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_close_fd(&fd);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic struct sql_db *driver_cassandra_init_v(const char *connect_string)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db = i_new(struct cassandra_db, 1);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->api = driver_cassandra_db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->fd_pipe[0] = db->fd_pipe[1] = -1;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen T_BEGIN {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen driver_cassandra_parse_connect_string(db, connect_string);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } T_END;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_log_set_level(db->log_level);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->timestamp_gen = cass_timestamp_gen_monotonic_new();

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->cluster = cass_cluster_new();

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_timestamp_gen(db->cluster, db->timestamp_gen);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_connect_timeout(db->cluster, db->connect_timeout_msecs);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_request_timeout(db->cluster, db->request_timeout_msecs);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_contact_points(db->cluster, db->hosts);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->user != NULL && db->password != NULL)

8576eb5abb66178f251c00209e564c7673c0e4cfTimo Sirainen cass_cluster_set_credentials(db->cluster, db->user, db->password);

975a784c2e02ecdcb56efb7a1db5e4769c7756d8Timo Sirainen if (db->port != 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_port(db->cluster, db->port);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->protocol_version != 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_protocol_version(db->cluster, db->protocol_version);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->num_threads != 0)

8576eb5abb66178f251c00209e564c7673c0e4cfTimo Sirainen cass_cluster_set_num_threads_io(db->cluster, db->num_threads);

8576eb5abb66178f251c00209e564c7673c0e4cfTimo Sirainen if (db->latency_aware_routing)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_latency_aware_routing(db->cluster, cass_true);

03936179f87aebde358dbe1ca8c34e5b5551db45Timo Sirainen if (db->heartbeat_interval_secs != 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_connection_heartbeat_interval(db->cluster, db->heartbeat_interval_secs);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->idle_timeout_secs != 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_set_connection_idle_timeout(db->cluster, db->idle_timeout_secs);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#ifdef HAVE_CASSANDRA_SPECULATIVE_POLICY

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->execution_retry_times > 0 && db->execution_retry_interval_msecs > 0)

3ea86ed7cf06ba04e4aa6cd1c4df9be336c06cd3Timo Sirainen cass_cluster_set_constant_speculative_execution_policy(db->cluster, db->execution_retry_interval_msecs, db->execution_retry_times);

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen#endif

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen db->session = cass_session_new();

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen if (db->metrics_path != NULL)

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen db->to_metrics = timeout_add(1000, driver_cassandra_metrics_write, db);

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen i_array_init(&db->results, 16);

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen i_array_init(&db->callbacks, 16);

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen return &db->api;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainenstatic void driver_cassandra_deinit_v(struct sql_db *_db)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen driver_cassandra_close(db, "Deinitialized");

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

3ea86ed7cf06ba04e4aa6cd1c4df9be336c06cd3Timo Sirainen i_assert(array_count(&db->callbacks) == 0);

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen array_free(&db->callbacks);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_assert(array_count(&db->results) == 0);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen array_free(&db->results);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_session_free(db->session);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_cluster_free(db->cluster);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen cass_timestamp_gen_free(db->timestamp_gen);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->to_metrics != NULL)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen timeout_remove(&db->to_metrics);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_free(db->metrics_path);

3ea86ed7cf06ba04e4aa6cd1c4df9be336c06cd3Timo Sirainen i_free(db->hosts);

3ea86ed7cf06ba04e4aa6cd1c4df9be336c06cd3Timo Sirainen i_free(db->error);

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen i_free(db->keyspace);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_free(db->user);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen i_free(db->password);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen array_free(&_db->module_contexts);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen i_free(db);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void driver_cassandra_result_unlink(struct cassandra_db *db,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_result *result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_result *const *results;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int i, count;

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen results = array_get(&db->results, &count);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen for (i = 0; i < count; i++) {

0c47c2096714b50880d48d00ce0bf28349eb4aceTimo Sirainen if (results[i] == result) {

0c47c2096714b50880d48d00ce0bf28349eb4aceTimo Sirainen array_delete(&db->results, i, 1);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen }

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen i_unreached();

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void driver_cassandra_log_result(struct cassandra_result *result,

a10ed8c47534b4c6b6bf2711ccfe577e720a47b4Timo Sirainen bool all_pages, long long reply_usecs)

a10ed8c47534b4c6b6bf2711ccfe577e720a47b4Timo Sirainen{

a10ed8c47534b4c6b6bf2711ccfe577e720a47b4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct timeval now;

471a6b2b4e64eca5d5779ae20a477312b32c89eeTimo Sirainen unsigned int row_count;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->log_level < CASS_LOG_DEBUG && !db->debug_queries &&

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen reply_usecs/1000 < db->warn_timeout_msecs)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (gettimeofday(&now, NULL) < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_fatal("gettimeofday() failed: %m");

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen string_t *str = t_str_new(128);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_printfa(str, "cassandra: Finished query '%s' (", result->query);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (all_pages) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen str_printfa(str, "%u pages in total, ", result->page_num);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen row_count = result->total_row_count;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen } else {

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen if (result->page_num > 0 || result->paging_continues)

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen str_printfa(str, "page %u, ", result->page_num);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen row_count = result->row_count;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen }

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen str_printfa(str, "%u rows, %lld+%lld us): %s", row_count, reply_usecs,

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen timeval_diff_usecs(&now, &result->finish_time),

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen result->error != NULL ? result->error : "success");

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen if (reply_usecs/1000 >= db->warn_timeout_msecs) {

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_SLOW]++;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen i_warning("%s", str_c(str));

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen } else {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_debug("%s", str_c(str));

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainenstatic void driver_cassandra_result_free(struct sql_result *_result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_result->db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_result *result = (struct cassandra_result *)_result;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen long long reply_usecs;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_assert(!result->api.callback);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_assert(result->callback == NULL);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (_result == db->sync_result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->sync_result = NULL;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen reply_usecs = timeval_diff_usecs(&result->finish_time, &result->start_time);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen driver_cassandra_log_result(result, FALSE, reply_usecs);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (result->page_num > 0 && !result->paging_continues) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* Multi-page query finishes now. Log a debug/warning summary

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen reply_usecs = timeval_diff_usecs(&result->finish_time,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen &result->page0_start_time);

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen driver_cassandra_log_result(result, TRUE, reply_usecs);

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen }

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen if (result->result != NULL)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_result_free(result->result);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (result->iterator != NULL)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_iterator_free(result->iterator);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (result->statement != NULL)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_statement_free(result->statement);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (result->row_pool != NULL)

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen pool_unref(&result->row_pool);

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen i_free(result->query);

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen i_free(result->error);

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen i_free(result);

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen}

34b8b14ce06c0939932b60f22f61aea124198438Timo Sirainen

34b8b14ce06c0939932b60f22f61aea124198438Timo Sirainenstatic void result_finish(struct cassandra_result *result)

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen{

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen bool free_result = TRUE;

d9b9687bf8cae9cfb070b1b7aadefa683220269fTimo Sirainen

9bbfe7f5ff821cac11d1d2550a91b148f389d82cTimo Sirainen result->finished = TRUE;

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen result->finish_time = ioloop_timeval;

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen driver_cassandra_result_unlink(db, result);

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen i_assert((result->error != NULL) == (result->iterator == NULL));

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen result->api.callback = TRUE;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen T_BEGIN {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->callback(&result->api, result->context);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } T_END;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen result->api.callback = FALSE;

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen free_result = db->sync_result != &result->api;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->ioloop != NULL)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen io_loop_stop(db->ioloop);

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_assert(!free_result || result->api.refcount > 0);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->callback = NULL;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (free_result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen sql_result_unref(&result->api);

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen}

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void query_resend_with_fallback(struct cassandra_result *result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen time_t last_warning =

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen ioloop_time - db->last_fallback_warning[result->query_type];

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen if (last_warning >= CASSANDRA_FALLBACK_WARN_INTERVAL_SECS) {

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen i_warning("%s - retrying future %s queries with consistency %s (instead of %s)",

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->error, cassandra_query_type_names[result->query_type],

2e86ce0239b4bc88c94d7edb5813d38ab080627eTimo Sirainen cass_consistency_string(result->fallback_consistency),

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_consistency_string(result->consistency));

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen db->last_fallback_warning[result->query_type] = ioloop_time;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_free_and_null(result->error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->fallback_failures[result->query_type]++;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->consistency = result->fallback_consistency;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen driver_cassandra_result_send_query(result);

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen}

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainenstatic void counters_inc_error(struct cassandra_db *db, CassError error)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen switch (error) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASS_ERROR_LIB_NO_HOSTS_AVAILABLE:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_NO_HOSTS]++;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASS_ERROR_LIB_REQUEST_QUEUE_FULL:

cf848255bf65a5e2382c59c093da72f877f7535aTimo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_QUEUE_FULL]++;

cf848255bf65a5e2382c59c093da72f877f7535aTimo Sirainen break;

cf848255bf65a5e2382c59c093da72f877f7535aTimo Sirainen case CASS_ERROR_LIB_REQUEST_TIMED_OUT:

cf848255bf65a5e2382c59c093da72f877f7535aTimo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_CLIENT_TIMEOUT]++;

cf848255bf65a5e2382c59c093da72f877f7535aTimo Sirainen break;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASS_ERROR_SERVER_WRITE_TIMEOUT:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_SERVER_TIMEOUT]++;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASS_ERROR_SERVER_UNAVAILABLE:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_SERVER_UNAVAILABLE]++;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen default:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_RECV_ERR_OTHER]++;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen break;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen }

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen}

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainenstatic void query_callback(CassFuture *future, void *context)

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen{

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen struct cassandra_result *result = context;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CassError error = cass_future_error_code(future);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (error != CASS_OK) {

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen const char *errmsg;

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen size_t errsize;

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen int msecs;

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_future_error_message(future, &errmsg, &errsize);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_free(result->error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen msecs = timeval_diff_msecs(&ioloop_timeval, &result->start_time);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen counters_inc_error(db, error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* Timeouts bring uncertainty whether the query succeeded or

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->api.error_type = error == CASS_ERROR_SERVER_WRITE_TIMEOUT ||

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen error == CASS_ERROR_SERVER_UNAVAILABLE ||

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen error == CASS_ERROR_LIB_REQUEST_TIMED_OUT ?

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen SQL_RESULT_ERROR_TYPE_WRITE_UNCERTAIN :

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen SQL_RESULT_ERROR_TYPE_UNKNOWN;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->error = i_strdup_printf("Query '%s' failed: %.*s (in %u.%03u secs%s)",

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->query, (int)errsize, errmsg, msecs/1000, msecs%1000,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->page_num == 0 ? "" : t_strdup_printf(", page %u", result->page_num));

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* unavailable = cassandra server knows that there aren't

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if ((error == CASS_ERROR_SERVER_UNAVAILABLE ||

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen error == CASS_ERROR_LIB_NO_HOSTS_AVAILABLE ||

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen error == CASS_ERROR_SERVER_WRITE_TIMEOUT) &&

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->fallback_consistency != result->consistency) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* retry with fallback consistency */

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen query_resend_with_fallback(result);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen result_finish(result);

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_RECV_OK]++;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (result->fallback_consistency != result->consistency) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* non-fallback query finished successfully. if there had been

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->fallback_failures[result->query_type] = 0;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->result = cass_future_get_result(future);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->iterator = cass_iterator_from_result(result->result);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result_finish(result);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void driver_cassandra_init_statement(struct cassandra_result *result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (result->statement != NULL) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* continuing a paged result */

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->statement = cass_statement_new(result->query, 0);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_statement_set_consistency(result->statement, result->consistency);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#ifdef HAVE_CASSANDRA_SPECULATIVE_POLICY

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_statement_set_is_idempotent(result->statement, cass_true);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen#endif

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->page_size > 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_statement_set_paging_size(result->statement, db->page_size);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void driver_cassandra_result_send_query(struct cassandra_result *result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CassFuture *future;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->counters[CASSANDRA_COUNTER_TYPE_QUERY_SENT]++;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen driver_cassandra_init_statement(result);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen future = cass_session_execute(db->session, result->statement);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen driver_cassandra_set_callback(future, db, query_callback, result);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic bool

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainendriver_cassandra_want_fallback_query(struct cassandra_result *result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int failure_count = db->fallback_failures[result->query_type];

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int i, msecs = CASSANDRA_FALLBACK_FIRST_RETRY_MSECS;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct timeval tv;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (failure_count == 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return FALSE;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* double the retries every time. */

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen for (i = 1; i < failure_count; i++) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen msecs *= 2;

252f5c6a63878e7a8a7ffb5847eecbad7f8737e8Timo Sirainen if (msecs >= CASSANDRA_FALLBACK_MAX_RETRY_MSECS) {

252f5c6a63878e7a8a7ffb5847eecbad7f8737e8Timo Sirainen msecs = CASSANDRA_FALLBACK_MAX_RETRY_MSECS;

252f5c6a63878e7a8a7ffb5847eecbad7f8737e8Timo Sirainen break;

252f5c6a63878e7a8a7ffb5847eecbad7f8737e8Timo Sirainen }

252f5c6a63878e7a8a7ffb5847eecbad7f8737e8Timo Sirainen }

252f5c6a63878e7a8a7ffb5847eecbad7f8737e8Timo Sirainen /* If last primary query sent timestamp + msecs is older than current

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen tv = db->primary_query_last_sent[result->query_type];

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen timeval_add_msecs(&tv, msecs);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return timeval_cmp(&ioloop_timeval, &tv) < 0;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic int driver_cassandra_send_query(struct cassandra_result *result)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen int ret;

252f5c6a63878e7a8a7ffb5847eecbad7f8737e8Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (!SQL_DB_IS_READY(&db->api)) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if ((ret = sql_connect(&db->api)) <= 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (ret < 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen driver_cassandra_close(db, "Couldn't connect to Cassandra");

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return ret;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (result->page0_start_time.tv_sec == 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->page0_start_time = ioloop_timeval;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->start_time = ioloop_timeval;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->row_pool = pool_alloconly_create("cassandra result", 512);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen switch (result->query_type) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASSANDRA_QUERY_TYPE_READ:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->consistency = db->read_consistency;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->fallback_consistency = db->read_fallback_consistency;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASSANDRA_QUERY_TYPE_READ_MORE:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* consistency is already set and we don't want to fallback

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->fallback_consistency = result->consistency;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASSANDRA_QUERY_TYPE_WRITE:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->consistency = db->write_consistency;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen result->fallback_consistency = db->write_fallback_consistency;

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen break;

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen case CASSANDRA_QUERY_TYPE_DELETE:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->consistency = db->delete_consistency;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->fallback_consistency = db->delete_fallback_consistency;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASSANDRA_QUERY_TYPE_COUNT:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_unreached();

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

59bf1df59552b6898749a87ae28476debd9b77d8Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (driver_cassandra_want_fallback_query(result))

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->consistency = result->fallback_consistency;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen else

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->primary_query_last_sent[result->query_type] = ioloop_timeval;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

31a574fda352ef4f71dbff9c30e15e4744e132c0Timo Sirainen driver_cassandra_result_send_query(result);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->query_sent = TRUE;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return 1;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void driver_cassandra_send_queries(struct cassandra_db *db)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

31a574fda352ef4f71dbff9c30e15e4744e132c0Timo Sirainen struct cassandra_result *const *results;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen unsigned int i, count;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen results = array_get(&db->results, &count);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen for (i = 0; i < count; i++) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (!results[i]->query_sent) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (driver_cassandra_send_query(results[i]) <= 0)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen }

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen }

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen}

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainenstatic void exec_callback(struct sql_result *_result ATTR_UNUSED,

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen void *context ATTR_UNUSED)

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen{

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen}

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainenstatic struct cassandra_result *

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainendriver_cassandra_query_init(struct cassandra_db *db, const char *query,

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen enum cassandra_query_type query_type,

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen sql_query_callback_t *callback, void *context)

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen{

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen struct cassandra_result *result;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen result = i_new(struct cassandra_result, 1);

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen result->api = driver_cassandra_result;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen result->api.db = &db->api;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen result->api.refcount = 1;

252f5c6a63878e7a8a7ffb5847eecbad7f8737e8Timo Sirainen result->callback = callback;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen result->context = context;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen result->query_type = query_type;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen result->query = i_strdup(query);

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen array_append(&db->results, &result, 1);

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen return result;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen}

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainenstatic void

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainendriver_cassandra_query_full(struct sql_db *_db, const char *query,

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen enum cassandra_query_type query_type,

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen sql_query_callback_t *callback, void *context)

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen{

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen struct cassandra_result *result;

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen

ccd7b4e0a5f09058a59cc4b3f878254e93e7cb0aTimo Sirainen result = driver_cassandra_query_init(db, query, query_type,

ccd7b4e0a5f09058a59cc4b3f878254e93e7cb0aTimo Sirainen callback, context);

ccd7b4e0a5f09058a59cc4b3f878254e93e7cb0aTimo Sirainen (void)driver_cassandra_send_query(result);

ccd7b4e0a5f09058a59cc4b3f878254e93e7cb0aTimo Sirainen}

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainenstatic void driver_cassandra_exec(struct sql_db *db, const char *query)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen driver_cassandra_query_full(db, query, CASSANDRA_QUERY_TYPE_WRITE, exec_callback, NULL);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic void driver_cassandra_query(struct sql_db *db, const char *query,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen sql_query_callback_t *callback, void *context)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen driver_cassandra_query_full(db, query, CASSANDRA_QUERY_TYPE_READ, callback, context);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainenstatic void cassandra_query_s_callback(struct sql_result *result, void *context)

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct cassandra_db *db = context;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->sync_result = result;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainenstatic void driver_cassandra_sync_init(struct cassandra_db *db)

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (sql_connect(&db->api) < 0)

cd70f7aec3bf49147fa80b77dd7ede7d7697202eTimo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->orig_ioloop = current_ioloop;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->ioloop = io_loop_create();

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (IS_CONNECTED(db))

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_assert(db->api.state == SQL_DB_STATE_CONNECTING);

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen db->io_pipe = io_loop_move_io(&db->io_pipe);

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen /* wait for connecting to finish */

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen io_loop_run(db->ioloop);

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen}

efeb13303798b47d2c4295468d233c1bcfd79c94Timo Sirainen

a75d470c9223a75801418fcdda258885c36317e0Timo Sirainenstatic void driver_cassandra_sync_deinit(struct cassandra_db *db)

a75d470c9223a75801418fcdda258885c36317e0Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->orig_ioloop == NULL)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (db->io_pipe != NULL) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen io_loop_set_current(db->orig_ioloop);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->io_pipe = io_loop_move_io(&db->io_pipe);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen io_loop_set_current(db->ioloop);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen io_loop_destroy(&db->ioloop);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic struct sql_result *

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainendriver_cassandra_sync_query(struct cassandra_db *db, const char *query)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen struct sql_result *result;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_assert(db->sync_result == NULL);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen switch (db->api.state) {

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen case SQL_DB_STATE_CONNECTING:

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen case SQL_DB_STATE_BUSY:

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen i_unreached();

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen case SQL_DB_STATE_DISCONNECTED:

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen sql_not_connected_result.refcount++;

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen return &sql_not_connected_result;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case SQL_DB_STATE_IDLE:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen driver_cassandra_query(&db->api, query, cassandra_query_s_callback, db);

6abf66a3731d52889517bd644595c540e3a9b3ecTimo Sirainen if (db->sync_result == NULL) {

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen db->io_pipe = io_loop_move_io(&db->io_pipe);

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen io_loop_run(db->ioloop);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result = db->sync_result;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (result == &sql_not_connected_result) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen /* we don't end up in cassandra's free function, so sync_result

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen db->sync_result = NULL;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen } else if (result == NULL) {

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen result = &sql_not_connected_result;

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen result->refcount++;

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen }

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen return result;

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen}

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainenstatic struct sql_result *

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainendriver_cassandra_query_s(struct sql_db *_db, const char *query)

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen{

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen struct sql_result *result;

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen

5446576156fbe26e07a5cb964a900281d283f387Timo Sirainen driver_cassandra_sync_init(db);

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen result = driver_cassandra_sync_query(db, query);

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen driver_cassandra_sync_deinit(db);

53c384be5f8f27945fd61b8a0d731a93a261628fTimo Sirainen return result;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen}

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainenstatic int

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainendriver_cassandra_get_value(struct cassandra_result *result,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen const CassValue *value, const char **str_r,

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen size_t *len_r)

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen{

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen const unsigned char *output;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen void *output_dup;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen size_t output_size;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen CassError rc;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen const char *type;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (cass_value_is_null(value) != 0) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen *str_r = NULL;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen return 0;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen switch (cass_data_type_type(cass_value_data_type(value))) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen case CASS_VALUE_TYPE_INT: {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_int32_t num;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen rc = cass_value_get_int32(value, &num);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (rc == CASS_OK) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen const char *str = t_strdup_printf("%d", num);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen output_size = strlen(str);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen output = (const void *)str;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen type = "int32";

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

0c47c2096714b50880d48d00ce0bf28349eb4aceTimo Sirainen }

0c47c2096714b50880d48d00ce0bf28349eb4aceTimo Sirainen case CASS_VALUE_TYPE_BIGINT: {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen cass_int64_t num;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen rc = cass_value_get_int64(value, &num);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen if (rc == CASS_OK) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen const char *str = t_strdup_printf("%lld", (long long)num);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen output_size = strlen(str);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen output = (const void *)str;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen type = "int64";

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen }

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen default:

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen rc = cass_value_get_bytes(value, &output, &output_size);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen type = "bytes";

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen break;

d8504a5ac72efb86a8da852c8d814ccaf0ae8625Timo Sirainen }

d8504a5ac72efb86a8da852c8d814ccaf0ae8625Timo Sirainen if (rc != CASS_OK) {

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen i_free(result->error);

7bd5b1c64cc987715bdaf8cc4907c3c37d5d7b29Timo Sirainen result->error = i_strdup_printf("Couldn't get value as %s: %s",