Pheromone
Pheromones are chemical substances secreted from glands and used as a means of communication.
pheromone allows setting up producers that publish ActiveRecord updates to Kafka whenever there is a model update and/or create.
Installation
Add this line to your application's Gemfile:
gem 'pheromone'And then execute:
$ bundle install
Or install it yourself as:
$ gem install pheromone
Pheromone Setup
Pheromone depends on waterdrop to send messages to Kafka. waterdrop settings can be added by following the Setup step on waterdrop
In order to setup pheromone, both waterdrop and pheromone need to be setup. Run this to generate pheromone configuration:
$ bundle exec rails generate pheromone:initializer
This will generate the following in config/initializers/pheromone.rb
Pheromone.setup do |pheromone_config|
# pheromone_config.background_processor.name = ':resque / :sidekiq'
# pheromone_config.background_processor.klass = 'BackgroundWorker'
# pheromone_config.timezone = 'UTC'
pheromone_config.message_format = :json
WaterDrop.setup do |waterdrop_config|
waterdrop_config.deliver = Rails.env.production?
waterdrop_config.kafka.seed_brokers = [Rails.env.production? ? ENV['KAFKA_HOST'] : 'localhost:9092']
end
end
Edit this file to modify the default config. The following configuration options are available:
| Option | Value type | Description |
|---|---|---|
| background_processor.name | Symbol | Choose :sidekiq or :resque as the background processor only if messages need to be sent to kafka asynchronously |
| background_processor.klass | String | Background processor class name that sends messages to kafka |
| background_processor.custom_processor | Proc | Custom processor (only works when you specify name as :custom) |
| timezone_format | String | Valid timezone name for timestamps sent to kafka |
| message_format | Symbol | Only supports :json format currently |
| enabled | Boolean | Defaults to true. When this is set to false, no messages will be sent to Kafka |
The timezone setting will transform any timestamp attributes in the message to the specified format.
Usage
1. Sending messages to kafka asynchronously
The underlying Kafka client used by pheromone is ruby-kafka. This client provides a normal producer that sends messages to Kafka synchronously, and an async_producer to send messages to Kafka asynchronously.
It is advisable to use the normal producer in production systems because async producer provides no guarantees that the messages will be delivered. To read more on this, refer the ruby-kafka documentation
Even while using a synchronous producer, sometimes there might be a need to run send messages to Kafka in a background task. This is especially true for batch processing tasks that send a high message volume to Kafka. To allow for this, pheromone provides an async mode that can be specified as an option to publish by specifying dispatch_method as :async. By default, dispatch_method will be :sync. Specifying :async will still use the normal producer and NOT the async_producer.
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:create],
topic: :topic_test,
message: ->(obj) { { name: obj.name } },
dispatch_method: :async
}
]
end
The background_processor can be set inside Pheromone.config.background_processor.name as either :resque or sidekiq.
1.a. Using :resque
Create a new class and add the name under Pheromone.config.background_processor.klass. Implement a class method perform(message), and invoke message.send! inside the method as shown below:
class ResqueJob
@queue = :low
def self.perform(message_object)
Pheromone::Messaging::Message.new(
topic: message_object['topic'],
metadata: { source: 'application1' },
blob: message_object['blob'],
metadata: message_object['metadata'],
options: message_object['options']
).send!
end
end
1.b. Using :sidekiq
Create a new class and add the name under Pheromone.config.background_processor.klass. Implement an instance method perform_async(message), and invoke message.send! inside the method as shown below:
class SidekiqJob
include Sidekiq::Worker
def perform(message_object)
Pheromone::Messaging::Message.new(
topic: message_object['topic'],
blob: message_object['blob'],
metadata: message_object['metadata'],
options: message_object['options']
).send!
end
end
1.c. Implement your own processor
You can also implement your own processor in addition to resque and sidekiq.
Pheromone.setup do |config|
config.background_processor.name = :custom
config.background_processor.klass = 'CustomJob'
config.background_processor.custom_processor = -> (klass, msg) do
klass.perform(msg)
end
endpheromone will invoke the class name specified in the config with the message object. This mode can be used if you don't want to block a request that ends up sending messages to Kafka.
2. Supported events
2.a. To send messages for model create event, add the following lines to your ActiveRecord model
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:create],
topic: :topic1,
metadata: { source: 'application1' },
message: ->(obj) { { name: obj.name } }
}
]
end
2.b. To send messages for model update event, specify update in the event_types array:
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:update],
topic: :topic1,
metadata: { source: 'application1' },
message: ->(obj) { { name: obj.name } }
}
]
end
Messages can be published for multiple event types by defining events_types: [:create, :update].
3. Supported message formats
3.a. Using a proc in message
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:create],
topic: :topic1,
metadata: { source: 'application1' },
message: ->(obj) { { name: obj.name } }
}
]
end
3.b. Using a defined function in message
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:update],
topic: :topic1,
metadata: { source: 'application1' },
message: message
}
]
def message
{ name: self.name }
end
end
3.c. Using a serializer in message
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:create],
topic: :topic1,
metadata: { source: 'application1' },
serializer: Class.new(BaseSerializer) { attributes :name, :type }
}
]
end
4. Sending messages conditionally
4.a. Using a proc in if
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:update],
topic: :topic1,
message: message,
if: ->(data) { data.condition },
}
]
def message
{ name: self.name }
end
end
4.b. Using a defined function in if
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:update],
topic: :topic1,
message: message,
if: pre_condition
}
]
def pre_condition
name.present?
end
def message
{ name: self.name }
end
end
5. Specifying the topic
The kafka topic can be specified in the topic option to publish. To publish to topic_test, use the following:
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:create],
topic: :topic_test,
message: ->(obj) { { name: obj.name } }
}
]
end
6. Specifying producer options
Ruby-Kafka allows sending options to change the behaviour of Kafka Producer.
These can be sent in by passing producer_options to the publish method:
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:create],
topic: :topic_test,
message: ->(obj) { { name: obj.name } },
producer_options: {
key: 'key',
partition: 1,
partition_key: 'key'
}
}
]
end
7. Specifying message metadata
Pheromone makes it easier to standardise the format of messages sent to Kafka while affording the flexibility to add other custom fields using the metadata option.
By default, the messages sent to Kafka have an event_type and timestamp attached to it as shown below:
{
'event_type' => 'create',
'timestamp' => '2015-07-14T10:10:00.000+08:00',
'blob' => {
'message_text' => 'test'
}
}.to_json
By specifying metadata to the publish method:
class PublishableModel < ActiveRecord::Base
include Pheromone::Publishable
publish [
{
event_types: [:create],
topic: :topic_test,
message: ->(obj) { { name: obj.name } },
metadata: { source: 'application1' },
producer_options: {
key: 'key',
partition: 1,
partition_key: 'key'
}
}
]
end
The Kafka message will have the original metadata in addition to the new fields:
{
'event_type' => 'create',
'timestamp' => '2015-07-14T10:10:00.000+08:00',
'source' => 'application1',
'blob' => {
'message_text' => 'test'
}
}.to_json
8. Sending messages to Kafka directly
pheromone provides a custom message object that sends messages to Kafka in a predefined format, to maintain consistency in the message fields.
Pheromone::Messaging::Message can be initialized with the following arguments:
-
topic: name of the topic to which the message is produced -
blob: the actual message itself -
metadata: any additional fields that must be sent along with the message -
options: producer options as described in Section 6
Of these fields, only topic and message are compulsory and the remaining two are optional.
Example usage:
Pheromone::Messaging::Message.new(
topic: 'test_topic',
blob: { message_text: 'test' },
metadata: { event_type: 'create' },
producer_options: { max_retries: 5 }
).send!
This will send a message to test_topic in Kafka in the following format:
{
'event_type' => 'create',
'timestamp' => '2015-07-14T10:10:00.000+08:00',
'blob' => {
'message_text' => 'test'
}
}.to_json
As seen above timestamp will be added automatically to the main attributes along with the message metadata. The actual message will be encapsulated inside a key called blob.
9. Encoding messages
pheromone allows encoding messages using custom encoders. Encoders can be specified by simply specifying a proc to encode the message in any given format.
publish takes the encoder and message_format as options for encoding to be specified. Encoder is called with the entire message object and performs encoding on this message.
publish(
[
{
topic: :test_topic,
event_types: [:update, :create],
serializer: TestSerializer,
message_format: :with_encoding,
encoder: ->(message) { avro.encode(message, schema_name: 'test') },
if: ->(object) { object.should_publish_status_update? }
}
]
)
Message Contents
In versions before < 0.5, the metadata keys were present at the top-most level, and the message itself was embedded inside the key blob.
The message published to Kafka looks like this:
{
event: 'create',
entity: 'KlassName',
timestamp: '2015-07-14T02:10:00.000Z',
blob: {
message_contents: 'message_contents'
}
}
From 0.5 version onwards, the message format looks like this:
{
metadata: {
event: 'create',
entity: 'KlassName',
timestamp: '2015-07-14T02:10:00.000Z'
}
blob: {
message_contents: 'message_contents'
}
}
event, entity, and timestamp are determined and added by pheromone. timestamp will be in UTC by default and will use timezone_format value specified in config/initializers/pheromone.rb. Contents of metadata can be modified by using these guidelines. Message contents are placed under the key blob.
In order to use the format with blob embedded inside metadata, specify option embed_blob as true inside publish options like this:
publish(
[
{
topic: :test_topic,
event_types: [:update, :create],
serializer: TestSerializer,
message_format: :with_encoding,
embed_blob: true
encoder: ->(message) { avro.encode(message, schema_name: 'test') },
if: ->(object) { object.should_publish_status_update? }
}
]
)
Testing
RSpec
pheromone makes it easy to control when it is enabled during testing with pheromone/frameworks/rspec
# spec/rails_helper.rb
require 'rspec/rails'
# ...
require 'pheromone/frameworks/rspec'
With this helper, publishable is disabled by default. To turn it on, pass publishable: true to the spec block like this:
describe 'PublishableModel' do
before do
@invocation_count = 0
allow(WaterDrop::SyncProducer).to receive(:call) do
@invocation_count += 1
end
it 'sends messages to kafka', publishable: true do
PublishableModel.create
expect(@invocation_count).to eq(1)
end
end
Alternatively, Pheromone.enabled? can be used to check if pheromone is enabled
Development
After checking out the repo, run bin/setup to install dependencies. You can also run bin/console for an interactive prompt that will allow you to experiment.
To install this gem onto your local machine, run bundle exec rake install. To release a new version, update the version number in version.rb, and then run bundle exec rake release, which will create a git tag for the version, push git commits and tags, and push the .gem file to rubygems.org.
Contributing
This project is intended to be a safe, welcoming space for collaboration, and contributors are expected to adhere to the Contributor Covenant code of conduct.
License
The gem is available as open source under the terms of the MIT License.
=======