google_cloud_pubsub.rb

# encoding: utf-8
require 'gcloud'

require 'faraday'
module Faraday
  class Adapter
    class NetHttp < Faraday::Adapter
      def ssl_verify_mode(ssl)
        OpenSSL::SSL::VERIFY_NONE
      end
    end
  end
end

require "logstash/outputs/base"
require "logstash/namespace"

# Stream events to a Google Cloud PubSub topic
class LogStash::Outputs::GoogleCloudPubSub < LogStash::Outputs::Base
  config_name "google_cloud_pubsub"

  config :project

  # Path to JSON file containing the Service Account credentials (not needed when running inside GCE)
  config :keyfile

  # The name of the PubSub topic
  config :topic, :validate => :string, :required => true

  # Autocreate the topic if it doesn't exist
  config :autocreate_topic, :validate => :boolean, :default => true

  # The name of the project that the topic is in (if it's not the current project)
  config :topic_project, :validate => :string

  # Maximum number of messages to pull in a single call
  config :batch_size, :validate => :number, :default => 10

  config :output_format, :validate => [ "json", "plain" ], :default => "plain"

  default :codec, "plain"

  public
  def register
    @logger.info("Registering Google Cloud PubSub output", :project => @project, :keyfile => @keyfile, :topic => @topic)

    @pubsub = Gcloud.new(project=@project, keyfile=@keyfile).pubsub
    @topic = @pubsub.topic(@topic, { :autocreate => @autocreate_topic, :project => @topic_project })

    raise "Topic #{@topic} not found" if not @topic
    @logger.debug("Topic: ", :topic => @topic)

  end # def register

  public
  def receive(event)
    puts event
    if (@output_format == "json")
      message = LogStash::Json.dump(event.to_hash)
    else
      message = event.to_s
    end
    sent = @topic.publish(message)
  end # def receive

end # class LogStash::Outputs::GCS