Talk to Insights via http

lib/travis/addons/handlers/insights.rb

@@ -1,6 +1,7 @@
 require 'travis/addons/handlers/base'
 require 'travis/addons/handlers/task'
-require 'travis/addons/model/broadcast'
+require 'travis/addons/support/insights'
+require 'travis/rollout'
 
 module Travis
   module Addons
@@ -20,12 +21,24 @@ def handle?
         end
 
         def handle
-          Travis::Sidekiq.insights(event, payload)
+          post? ? post : enqueue
         end
 
         private
 
-          def payload
+          def post?
+            Rollout.matches?(:insights_http, owner: owner_name)
+          end
+
+          def post
+            insights.post(event: event, data: data)
+          end
+
+          def enqueue
+            Travis::Sidekiq.insights(event, data)
+          end
+
+          def data
             {
               type: object.class.name,
               id: object.id,
@@ -39,6 +52,7 @@ def payload
               finished_at: object.finished_at
             }
           end
+          alias payload data
 
           def state
             case event
@@ -48,6 +62,14 @@ def state
             end
           end
 
+          def owner_name
+            object.owner&.login
+          end
+
+          def insights
+            Travis::Insights.new(Hub.context.config)
+          end
+
           class EventHandler < Addons::Instrument
             def notify_completed
               publish

lib/travis/addons/support/insights.rb

@@ -0,0 +1,41 @@
+require 'faraday'
+
+module Travis
+  class Insights < Struct.new(:config)
+    def post(data)
+      http.post do |r|
+        r.url '/events'
+        r.params['source'] = 'hub'
+        r.headers['Content-Type'] = 'application/json'
+        r.body = JSON.dump(data)
+      end
+    end
+
+    private
+
+      def http
+        @http ||= Faraday.new(compact(url: url, ssl: ssl)) do |c|
+          c.request :authorization, :Token, %(token="#{token}")
+          c.request :retry, max: 5, methods: [:post], exceptions: [Faraday::Error]
+          c.response :raise_error
+          c.adapter :net_http
+        end
+      end
+
+      def url
+        config[:insights][:url]
+      end
+
+      def token
+        config[:insights][:token]
+      end
+
+      def ssl
+        config[:ssl].to_h
+      end
+
+      def compact(hash)
+        hash.reject { |_, value| value.nil? || value&.empty? }
+      end
+  end
+end

lib/travis/hub/config.rb

@@ -18,6 +18,7 @@ def http_basic_auth
              database:       { adapter: 'postgresql', database: "travis_#{env}", encoding: 'unicode', min_messages: 'warning', pool: 25, reaping_frequency: 60, variables: { statement_timeout: 10000 } },
              logs_api:       { url: 'https://travis-logs-notset.example.com:1234', token: 'notset', retries: { max: 5, interval: 3, max_interval: 60, interval_randomness: 0.5, backoff_factor: 2 } },
              job_board:      { url: 'https://not:[email protected]', site: 'org' },
+             insights:       { url: ENV['INSIGHTS_URL'], token: ENV['INSIGHTS_TOKEN'] },
              redis:          { url: ENV['TRAVIS_REDIS_URL'] || 'redis://localhost:6379', insights_url: ENV['INSIGHTS_REDIS_URL'] || 'redis://localhost:6379' },
              sidekiq:        { namespace: 'sidekiq', pool_size: 1 },
              lock:           { strategy: :redis, ttl: 30000 },
@@ -34,6 +35,7 @@ def http_basic_auth
              limit:          { resets: { max: 50, after: 6 * 60 * 60 } },
              notifications:  [],
              auth:           { jwt_public_key: ENV['JWT_RSA_PUBLIC_KEY'], http_basic_auth: http_basic_auth }
+
       def metrics
         # TODO cleanup keychain?
         super.to_h.merge(librato: librato.to_h.merge(source: librato_source), graphite: graphite)

spec/spec_helper.rb

@@ -6,6 +6,7 @@
 require 'travis/hub/api'
 require 'date'
 
+require 'support/config'
 require 'support/context'
 require 'support/database_cleaner'
 require 'support/env'
@@ -28,6 +29,7 @@
 RSpec.configure do |c|
   c.mock_with :mocha
   c.filter_run_excluding pending: true
+  c.include Support::Config
   c.include Support::Context
   c.include Support::DatabaseCleaner
   c.include Support::Env

spec/support/config.rb

@@ -0,0 +1,23 @@
+module Support
+  module Config
+    def self.included(base)
+      base.send(:extend, ClassMethods)
+    end
+
+    module ClassMethods
+      def config(vars)
+        before { define_config(vars) }
+        after  { undefine_config(vars) }
+      end
+    end
+
+    def define_config(vars)
+      vars.each { |key, value| context.config[key] = value }
+    end
+
+    def undefine_config(vars)
+      vars.each { |key, _| context.config[key] = nil }
+    end
+  end
+end
+

spec/travis/addons/handlers/insights_spec.rb

@@ -1,6 +1,7 @@
 describe Travis::Addons::Handlers::Insights do
-  let(:build)   { FactoryGirl.create(:build, repository: repository) }
-  let(:job)     { FactoryGirl.create(:job, owner_id: 1, owner_type: 'User', repository: repository) }
+  let(:owner)   { FactoryGirl.create(:user, login: 'user') }
+  let(:build)   { FactoryGirl.create(:build, owner: owner, repository: repository) }
+  let(:job)     { FactoryGirl.create(:job, owner: owner, repository: repository) }
   let(:repository) { FactoryGirl.create(:repository) }
 
   describe 'subscription' do
@@ -69,7 +70,7 @@
     }
   end
 
-  describe 'job:created' do
+  describe 'job:created via Sidekiq' do
     let(:event) { 'job:created' }
     it 'handle' do
       ::Sidekiq::Client.any_instance.expects(:push).with(
@@ -82,6 +83,24 @@
     end
   end
 
+  describe 'job:created via HTTP' do
+    config insights: { url: 'https://insights.travis-ci.com', token: 'token' }
+
+    env ROLLOUT: 'insights_http'
+    env ROLLOUT_INSIGHTS_HTTP_OWNERS: 'user'
+
+    let(:event) { 'job:created' }
+
+    it 'handle' do
+      stub_request(:post, 'https://insights.travis-ci.com/events?source=hub').with do |r|
+        expect(JSON.parse(r.body)['event']).to eq 'job:created'
+        expect(r.headers['Authorization']).to eq 'Token token="token"'
+        expect(r.headers['Content-Type']).to eq 'application/json'
+      end
+      handler.handle
+    end
+  end
+
   describe 'job:canceled' do
     let(:event) { 'job:canceled' }
     it 'handle' do