pronix/fluent-plugin-cql-rb

fluentd plugin for cassandra output via CQL 3.0

Cassandra plugin for Fluentd

Cassandra output plugin for Fluentd.

Implemented using the cql-rb gem and targets CQL 3.0.0 and Cassandra 2.x.x

Raison d'être

Currently, there's another Fluentd Cassandra plugin

It's implemented via the Twitter Cassandra gem, which:

a) doesn't provide all of the niceties of CQL, i.e., create/alter/delete keyspaces/columnfamilies
b) doesn't allow a desktop client to make a call to a Cassandra instance hosted on EC2
   (the gem resolves a cassandra node's IP address to its private EC2
    IP address (ex: 10.x.x.x), which isn't accessible outside EC2)

Installation

via RubyGems

fluent-gem install fluent-plugin-cql-rb

Quick Start

Cassandra Configuration

# create keyspace (via CQL)
  CREATE KEYSPACE "FluentdLoggers" WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 3 } ;

# switch to actual keyspace
   USE "FluentdLoggers";

# create table (column family)
  CREATE TABLE events (id varchar, ts bigint, payload text, PRIMARY KEY (id, ts)) WITH CLUSTERING ORDER BY (ts DESC);

# NOTE: schema definition should match that specified in the Fluentd.conf configuration file (see below)

Fluentd.conf Configuration

<match cql_rb.**>
  type cql         # fluent output plugin file name (sans fluent_plugin_ prefix)
  host 127.0.0.1             # cassandra hostname.
  port 9160                  # cassandra thrift port.
  keyspace FluentdLoggers    # cassandra keyspace
  columnfamily spec_events   # cassandra column family
  ttl 60                     # cassandra ttl *optional => default is 0*
  schema  {:id => :string, :ts => :bigint, :payload => :string}                   # cassandra column family schema *hash where keys => column names and values => data types*
  data_keys                  # comma delimited string of the fluentd hash's keys
  pop_data_keys              # keep or pop key/values from the fluentd hash when storing it as json
</match>

Tests

```
curl -X POST -d 'json={"id":1, "ts": 1, "payload": {"action":"login","user":2}}' http://127.0.0.1:8888/test.tag.here
```

rake rspec

NOTE: requires that cassandra be installed on the machine running the
      test as well as a keyspace named "FluentdLoggers" and a column family
      named "spec_events"

TODOs

1) make host and port configurable for tests
2) add rake task to generate keyspace and columnfamily
3) support multiple ip addresses in the connection string for Cassandra multi-node environments
4) make the cql version configurable

fluent-plugin-cql-rb's Issues

error parsing config

fluentd -c ./fluent/fluent.conf -vv
2015-01-14 09:48:25 +0100 [info]: fluent/supervisor.rb:341:read_config: reading config file path="./fluent/fluent.conf"
/var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/config/basic_parser.rb:88:in `parse_error!': got incomplete JSON hash configuration at fluent.conf line 28,1 (Fluent::ConfigParseError)
27:
28:

 -^
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/config/literal_parser.rb:236:in `scan_json'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/config/literal_parser.rb:59:in `parse_literal'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/config/v1_parser.rb:128:in `parse_element'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/config/v1_parser.rb:93:in `parse_element'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/config/v1_parser.rb:41:in `parse!'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/config/v1_parser.rb:31:in `parse'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/config.rb:29:in `parse'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/supervisor.rb:350:in `read_config'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/supervisor.rb:126:in `start'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/lib/fluent/command/fluentd.rb:167:in `<top (required)>'
    from /usr/lib/ruby/2.1.0/rubygems/core_ext/kernel_require.rb:73:in `require'
    from /usr/lib/ruby/2.1.0/rubygems/core_ext/kernel_require.rb:73:in `require'
    from /var/lib/gems/2.1.0/gems/fluentd-0.12.2/bin/fluentd:6:in `<top (required)>'
    from /usr/local/bin/fluentd:23:in `load'
    from /usr/local/bin/fluentd:23:in `<main>'

type syslog port 24224 tag rsyslog

<match rsyslog.**>
type copy

# for debug (see /var/log/td-agent.log)
type stdout

<match cql_rb.**>
type cql # fluent output plugin file name (sans fluent_plugin_ prefix)
keyspace FluentdLoggers # cassandra keyspace
columnfamily spec_events # cassandra column family
ttl 60 # cassandra ttl optional => default is 0
schema {:id => :string, :ts => :bigint, :payload => :string} # cassandra column family schema hash where keys => column names and values => data types
data_keys # comma delimited string of the fluentd hash's keys
pop_data_keys # keep or pop key/values from the fluentd hash when storing it as json

host 10.10.10.51 # cassandra's hostname. default localhost
port 9160 # cassandra's thrft port. default 9160