Write to InfluxDB directly via UDP

This removes the need for Sidekiq and any overhead/problems introduced
by TCP. There are a few things to take into account:

1. When writing data to InfluxDB you may still get an error if the
   server becomes unavailable during the write. Because of this we're
   catching all exceptions and just ignore them (for now).
2. Writing via UDP apparently requires the timestamp to be in
   nanoseconds. Without this data either isn't written properly.
3. Due to the restrictions on UDP buffer sizes we're writing metrics one
   by one, instead of writing all of them at once.

Procfile

@@ -3,5 +3,5 @@
 # lib/support/init.d, which call scripts in bin/ .
 #
 web: bundle exec unicorn_rails -p ${PORT:="3000"} -E ${RAILS_ENV:="development"} -c ${UNICORN_CONFIG:="config/unicorn.rb"}
-worker: bundle exec sidekiq -q post_receive -q mailers -q archive_repo -q system_hook -q project_web_hook -q gitlab_shell -q incoming_email -q runner -q common -q default -q metrics
+worker: bundle exec sidekiq -q post_receive -q mailers -q archive_repo -q system_hook -q project_web_hook -q gitlab_shell -q incoming_email -q runner -q common -q default
 # mail_room: bundle exec mail_room -q -c config/mail_room.yml

app/controllers/admin/application_settings_controller.rb

@@ -69,7 +69,7 @@ class Admin::ApplicationSettingsController < Admin::ApplicationController
       :max_artifacts_size,
       :metrics_enabled,
       :metrics_host,
-      :metrics_database,
+      :metrics_port,
       :metrics_username,
       :metrics_password,
       :metrics_pool_size,

app/views/admin/application_settings/_form.html.haml

@@ -171,12 +171,14 @@
       .col-sm-10
         = f.text_field :metrics_host, class: 'form-control', placeholder: 'influxdb.example.com'
     .form-group
-      = f.label :metrics_database, 'InfluxDB database', class: 'control-label col-sm-2'
+      = f.label :metrics_port, 'InfluxDB port', class: 'control-label col-sm-2'
       .col-sm-10
-        = f.text_field :metrics_database, class: 'form-control', placeholder: 'gitlab'
+        = f.text_field :metrics_port, class: 'form-control', placeholder: '8089'
         .help-block
-          The name of the InfluxDB database to store data in. Users will have to
-          create this database manually, GitLab does not do so automatically.
+          The UDP port to use for connecting to InfluxDB. InfluxDB requires that
+          your server configuration specifies a database to store data in when
+          sending messages to this port, without it metrics data will not be
+          saved.
     .form-group
       = f.label :metrics_username, 'InfluxDB username', class: 'control-label col-sm-2'
       .col-sm-10

app/workers/metrics_worker.rb

-class MetricsWorker
-  include Sidekiq::Worker
-
-  sidekiq_options queue: :metrics
-
-  def perform(metrics)
-    prepared = prepare_metrics(metrics)
-
-    Gitlab::Metrics.pool.with do |connection|
-      connection.write_points(prepared)
-    end
-  end
-
-  def prepare_metrics(metrics)
-    metrics.map do |hash|
-      new_hash = hash.symbolize_keys
-
-      new_hash[:tags].each do |key, value|
-        if value.blank?
-          new_hash[:tags].delete(key)
-        else
-          new_hash[:tags][key] = escape_value(value)
-        end
-      end
-
-      new_hash
-    end
-  end
-
-  def escape_value(value)
-    value.to_s.gsub('=', '\\=')
-  end
-end

db/migrate/20151229102248_influxdb_udp_port_setting.rb

+class InfluxdbUdpPortSetting < ActiveRecord::Migration
+  def change
+    add_column :application_settings, :metrics_port, :integer, default: 8089
+  end
+end

db/migrate/20151229112614_influxdb_remote_database_setting.rb

+class InfluxdbRemoteDatabaseSetting < ActiveRecord::Migration
+  def change
+    remove_column :application_settings, :metrics_database
+  end
+end

db/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 20151228175719) do
+ActiveRecord::Schema.define(version: 20151229112614) do
 
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
@@ -54,7 +54,6 @@ ActiveRecord::Schema.define(version: 20151228175719) do
     t.integer  "two_factor_grace_period",           default: 48
     t.boolean  "metrics_enabled",                   default: false
     t.string   "metrics_host",                      default: "localhost"
-    t.string   "metrics_database",                              default: "gitlab"
     t.string   "metrics_username"
     t.string   "metrics_password"
     t.integer  "metrics_pool_size",                 default: 16
@@ -63,6 +62,7 @@ ActiveRecord::Schema.define(version: 20151228175719) do
     t.boolean  "recaptcha_enabled",                 default: false
     t.string   "recaptcha_site_key"
     t.string   "recaptcha_private_key"
+    t.integer  "metrics_port",                      default: 8089
   end
 
   create_table "audit_events", force: :cascade do |t|

lib/gitlab/metrics.rb

@@ -66,6 +66,39 @@ module Gitlab
       end
     end
 
+    def self.submit_metrics(metrics)
+      prepared = prepare_metrics(metrics)
+
+      pool.with do |connection|
+        prepared.each do |metric|
+          begin
+            connection.write_points([metric])
+          rescue StandardError
+          end
+        end
+      end
+    end
+
+    def self.prepare_metrics(metrics)
+      metrics.map do |hash|
+        new_hash = hash.symbolize_keys
+
+        new_hash[:tags].each do |key, value|
+          if value.blank?
+            new_hash[:tags].delete(key)
+          else
+            new_hash[:tags][key] = escape_value(value)
+          end
+        end
+
+        new_hash
+      end
+    end
+
+    def self.escape_value(value)
+      value.to_s.gsub('=', '\\=')
+    end
+
     @hostname = Socket.gethostname
 
     # When enabled this should be set before being used as the usual pattern
@@ -73,11 +106,12 @@ module Gitlab
     if enabled?
       @pool = ConnectionPool.new(size: pool_size, timeout: timeout) do
         host = settings[:metrics_host]
-        db   = settings[:metrics_database]
         user = settings[:metrics_username]
         pw   = settings[:metrics_password]
+        port = settings[:metrics_port]
 
-        InfluxDB::Client.new(db, host: host, username: user, password: pw)
+        InfluxDB::Client.
+          new(udp: { host: host, port: port }, username: user, password: pw)
       end
     end
   end

lib/gitlab/metrics/metric.rb

@@ -26,7 +26,7 @@ module Gitlab
             process_type:   Sidekiq.server? ? 'sidekiq' : 'rails'
           ),
           values:    @values,
-          timestamp: @created_at.to_i
+          timestamp: @created_at.to_i * 1_000_000_000
         }
       end
     end

lib/gitlab/metrics/obfuscated_sql.rb

@@ -40,7 +40,7 @@ module Gitlab
           sql = sql.delete('"')
         end
 
-        sql.gsub("\n", ' ')
+        sql.tr("\n", ' ')
       end
     end
   end

lib/gitlab/metrics/sampler.rb

@@ -46,7 +46,7 @@ module Gitlab
       end
 
       def flush
-        MetricsWorker.perform_async(@metrics.map(&:to_hash))
+        Metrics.submit_metrics(@metrics.map(&:to_hash))
       end
 
       def sample_memory_usage

lib/gitlab/metrics/sidekiq_middleware.rb

@@ -5,13 +5,6 @@ module Gitlab
     # This middleware is intended to be used as a server-side middleware.
     class SidekiqMiddleware
       def call(worker, message, queue)
-        # We don't want to track the MetricsWorker itself as otherwise we'll end
-        # up in an infinite loop.
-        if worker.class == MetricsWorker
-          yield
-          return
-        end
-
         trans = Transaction.new
 
         begin

lib/gitlab/metrics/transaction.rb

@@ -59,7 +59,7 @@ module Gitlab
       end
 
       def submit
-        MetricsWorker.perform_async(@metrics.map(&:to_hash))
+        Metrics.submit_metrics(@metrics.map(&:to_hash))
       end
     end
   end

spec/lib/gitlab/metrics/sampler_spec.rb

@@ -38,7 +38,7 @@ describe Gitlab::Metrics::Sampler do
 
   describe '#flush' do
     it 'schedules the metrics using Sidekiq' do
-      expect(MetricsWorker).to receive(:perform_async).
+      expect(Gitlab::Metrics).to receive(:submit_metrics).
         with([an_instance_of(Hash)])
 
       sampler.sample_memory_usage

spec/lib/gitlab/metrics/sidekiq_middleware_spec.rb

@@ -11,14 +11,6 @@ describe Gitlab::Metrics::SidekiqMiddleware do
 
       middleware.call(worker, 'test', :test) { nil }
     end
-
-    it 'does not track jobs of the MetricsWorker' do
-      worker = MetricsWorker.new
-
-      expect(Gitlab::Metrics::Transaction).to_not receive(:new)
-
-      middleware.call(worker, 'test', :test) { nil }
-    end
   end
 
   describe '#tag_worker' do

spec/lib/gitlab/metrics/transaction_spec.rb

@@ -68,7 +68,7 @@ describe Gitlab::Metrics::Transaction do
     it 'submits the metrics to Sidekiq' do
       transaction.track_self
 
-      expect(MetricsWorker).to receive(:perform_async).
+      expect(Gitlab::Metrics).to receive(:submit_metrics).
         with([an_instance_of(Hash)])
 
       transaction.submit

spec/lib/gitlab/metrics_spec.rb

@@ -33,4 +33,52 @@ describe Gitlab::Metrics do
       expect(file).to eq('spec/lib/gitlab/metrics_spec.rb')
     end
   end
+
+  describe '#submit_metrics' do
+    it 'prepares and writes the metrics to InfluxDB' do
+      connection = double(:connection)
+      pool       = double(:pool)
+
+      expect(pool).to receive(:with).and_yield(connection)
+      expect(connection).to receive(:write_points).with(an_instance_of(Array))
+      expect(Gitlab::Metrics).to receive(:pool).and_return(pool)
+
+      described_class.submit_metrics([{ 'series' => 'kittens', 'tags' => {} }])
+    end
+  end
+
+  describe '#prepare_metrics' do
+    it 'returns a Hash with the keys as Symbols' do
+      metrics = described_class.
+        prepare_metrics([{ 'values' => {}, 'tags' => {} }])
+
+      expect(metrics).to eq([{ values: {}, tags: {} }])
+    end
+
+    it 'escapes tag values' do
+      metrics = described_class.prepare_metrics([
+        { 'values' => {}, 'tags' => { 'foo' => 'bar=' } }
+      ])
+
+      expect(metrics).to eq([{ values: {}, tags: { 'foo' => 'bar\\=' } }])
+    end
+
+    it 'drops empty tags' do
+      metrics = described_class.prepare_metrics([
+        { 'values' => {}, 'tags' => { 'cats' => '', 'dogs' => nil } }
+      ])
+
+      expect(metrics).to eq([{ values: {}, tags: {} }])
+    end
+  end
+
+  describe '#escape_value' do
+    it 'escapes an equals sign' do
+      expect(described_class.escape_value('foo=')).to eq('foo\\=')
+    end
+
+    it 'casts values to Strings' do
+      expect(described_class.escape_value(10)).to eq('10')
+    end
+  end
 end

spec/workers/metrics_worker_spec.rb

-require 'spec_helper'
-
-describe MetricsWorker do
-  let(:worker) { described_class.new }
-
-  describe '#perform' do
-    it 'prepares and writes the metrics to InfluxDB' do
-      connection = double(:connection)
-      pool       = double(:pool)
-
-      expect(pool).to receive(:with).and_yield(connection)
-      expect(connection).to receive(:write_points).with(an_instance_of(Array))
-      expect(Gitlab::Metrics).to receive(:pool).and_return(pool)
-
-      worker.perform([{ 'series' => 'kittens', 'tags' => {} }])
-    end
-  end
-
-  describe '#prepare_metrics' do
-    it 'returns a Hash with the keys as Symbols' do
-      metrics = worker.prepare_metrics([{ 'values' => {}, 'tags' => {} }])
-
-      expect(metrics).to eq([{ values: {}, tags: {} }])
-    end
-
-    it 'escapes tag values' do
-      metrics = worker.prepare_metrics([
-        { 'values' => {}, 'tags' => { 'foo' => 'bar=' } }
-      ])
-
-      expect(metrics).to eq([{ values: {}, tags: { 'foo' => 'bar\\=' } }])
-    end
-
-    it 'drops empty tags' do
-      metrics = worker.prepare_metrics([
-        { 'values' => {}, 'tags' => { 'cats' => '', 'dogs' => nil } }
-      ])
-
-      expect(metrics).to eq([{ values: {}, tags: {} }])
-    end
-  end
-
-  describe '#escape_value' do
-    it 'escapes an equals sign' do
-      expect(worker.escape_value('foo=')).to eq('foo\\=')
-    end
-
-    it 'casts values to Strings' do
-      expect(worker.escape_value(10)).to eq('10')
-    end
-  end
-end