Initiail code, extracted as open source

.gitignore

@@ -0,0 +1,17 @@
+*.gem
+*.rbc
+.bundle
+.config
+.yardoc
+Gemfile.lock
+InstalledFiles
+_yardoc
+coverage
+doc/
+lib/bundler/man
+pkg
+rdoc
+spec/reports
+test/tmp
+test/version_tmp
+tmp

Gemfile

@@ -0,0 +1,4 @@
+source 'https://rubygems.org'
+
+# Specify your gem's dependencies in kage.gemspec
+gemspec

LICENSE

@@ -0,0 +1,22 @@
+Copyright (c) 2012 Tatsuhiko Miyagawa
+
+MIT License
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,91 @@
+# Kage
+
+Kage is an HTTP shadow proxy server that sits between clients and your server(s) to enable "shadow requests".
+
+Kage can be used to duplex requests to the master (production) server and shadow servers that have newer code changes that are going to be deployed. By shadowing requests to the new code you can make sure there are no big/surprising changes in the response in terms of data, performance and database loads etc.
+
+You can customize the behavior of Kage with simple callbacks, when it chooses which backends to send shadow requests to (or not at all), appends or deletes HTTP headers per backend, and examines the complete HTTP response (including headers and body).
+
+## Features
+
+* Support HTTP/1.0 and HTTP/1.1 with partial keep-alive support (See below)
+* Callback to decide backends per request URLs
+* Callback to manipulate request headers per request and backend
+* Callback to examine responses from multiple backends (e.g. calcurate diffs)
+
+Kage does not yet support:
+
+* SSL
+* HTTP/1.1 requests pipelining
+
+## Usage
+
+```ruby
+require 'kage'
+
+def compare(a, b)
+  p [a, b]
+end
+
+Kage::ProxyServer.start do |server|
+  server.port = 8090
+  server.host = '0.0.0.0'
+  server.debug = false
+
+  # backends can share the same host/port
+  server.add_master_backend(:production, 'localhost', 80)
+  server.add_backend(:sandbox, 'localhost', 80)
+
+  server.client_timeout = 15
+  server.backend_timeout = 10
+
+  # Dispatch all GET requests to multiple backends, otherwise only :production
+  server.on_select_backends do |request, headers|
+    if request[:method] == 'GET'
+      [:production, :sandbox]
+    else
+      [:production]
+    end
+  end
+
+  # Add optional headers
+  server.on_munge_headers do |backend, headers|
+    headers['X-Kage-Session'] = self.session_id
+    headers['X-Kage-Sandbox'] = 1 if backend == :sandbox
+  end
+
+  # This callback is only fired when there are multiple backends to respond
+  server.on_backends_finished do |backends, requests, responses|
+    compare(responses[:production][:data], responses[:sandbox][:data])
+  end
+end
+```
+
+Read more sample code under the `examples/` directory.
+
+## Keep-alives
+
+Kage supports keep-alives for single backend requests, i.e. for requests where `on_select_backends` returns only the master backend.
+
+To make `on_backend_finished` callback simpler, if the current request matches with multiple backends, Kage sends `Connection: close` to the backends so that the callback will only get one response per backend in `responses` hash, which would look like:
+
+```ruby
+responses = {
+  :original => {:data => "(RAW HTTP response)", :elapsed => 0.1234},
+  :sandbox  => {:data => "(RAW HTTP response)", :elspaed => 0.2333},
+}
+```
+
+## Authors
+
+Tatsuhiko Miyagawa, Yusuke Mito
+
+## Acknowledgements
+
+Ilya Grigorik, Jos Boumans
+
+## Based On
+
+* [EventMachine](http://rubyeventmachine.com/)
+* [em-proxy](https://github.com/igrigorik/em-proxy/)
+* [http_parser.rb](https://github.com/tmm1/http_parser.rb)

Rakefile

@@ -0,0 +1,2 @@
+#!/usr/bin/env rake
+require "bundler/gem_tasks"

examples/proxy.rb

@@ -0,0 +1,39 @@
+#!/usr/bin/env ruby
+require 'kage'
+
+def compare(a, b)
+  p [a, b]
+end
+
+Kage::ProxyServer.start do |server|
+  server.port = 8090
+  server.host = '0.0.0.0'
+  server.debug = false
+
+  # backends can share the same host/port
+  server.add_master_backend(:production, 'localhost', 80)
+  server.add_backend(:sandbox, 'localhost', 80)
+
+  server.client_timeout = 15
+  server.backend_timeout = 10
+
+  # Dispatch all GET requests to multiple backends, otherwise only :production
+  server.on_select_backends do |request, headers|
+    if request[:method] == 'GET'
+      [:production, :sandbox]
+    else
+      [:production]
+    end
+  end
+
+  # Add optional headers
+  server.on_munge_headers do |backend, headers|
+    headers['X-Kage-Session'] = self.session_id
+    headers['X-Kage-Sandbox'] = 1 if backend == :sandbox
+  end
+
+  # This callback is only fired when there are multiple backends to respond
+  server.on_backends_finished do |backends, requests, responses|
+    compare(responses[:production][:data], responses[:sandbox][:data])
+  end
+end

kage.gemspec

@@ -0,0 +1,22 @@
+# -*- encoding: utf-8 -*-
+require File.expand_path('../lib/kage/version', __FILE__)
+
+Gem::Specification.new do |gem|
+  gem.authors       = ["Tatsuhiko Miyagawa"]
+  gem.email         = ["miyagawa@bulknews.net"]
+  gem.description   = %q{em-proxy based shadow proxy server}
+  gem.summary       = %q{Kage (kah-geh) is an HTTP shadow proxy server that sits between your production servers to send shaddow traffic to the servers with new code changes.}
+  gem.homepage      = "https://github.com/cookpad/kage"
+
+  gem.files         = `git ls-files`.split($\)
+  gem.executables   = gem.files.grep(%r{^bin/}).map{ |f| File.basename(f) }
+  gem.test_files    = gem.files.grep(%r{^(test|spec|features)/})
+  gem.name          = "kage"
+  gem.require_paths = ["lib"]
+  gem.version       = Kage::VERSION
+
+  gem.add_dependency 'em-proxy', '>= 0.1.7'
+  gem.add_dependency 'http_parser.rb', '>= 0.5.3'
+
+  gem.add_development_dependency 'rspec'
+end

lib/kage.rb

@@ -0,0 +1,3 @@
+require "kage/version"
+require "kage/connection"
+require "kage/proxy_server"

lib/kage/connection.rb

@@ -0,0 +1,167 @@
+require 'http/parser'
+
+module Kage
+  module Connection
+    attr_accessor :master_backend, :session_id
+
+    # http://eventmachine.rubyforge.org/EventMachine/Connection.html#unbind-instance_method
+    def close_connection(*args)
+      @server_side_close = true
+      super
+    end
+
+    def unbind
+      if !@server_side_close
+        if @state == :request
+          info "Client disconnected in the request phase"
+          super
+        elsif @backends && @backends.size == 1 && @responses[master_backend]
+          info "Client disconnected after the master response. Closing master"
+          super
+        else
+          info "Client disconnected. Waiting for all backends to finish"
+        end
+      end
+    end
+
+    def cleanup!
+      @parser.reset!
+    end
+
+    def all_servers_finished?
+      @servers.values.compact.size.zero?
+    end
+
+    def callback(cb, *args)
+      if @callbacks[cb]
+        instance_exec *args, &@callbacks[cb]
+      elsif block_given?
+        yield
+      end
+    rescue Exception => e
+      info "#{e} - #{e.backtrace}"
+    end
+
+    def build_headers(parser, headers)
+      "#{parser.http_method} #{parser.request_url} HTTP/#{parser.http_version.join(".")}\r\n" +
+      headers.map{|k, v| "#{k}: #{v}\r\n" }.join('') +
+      "\r\n"
+    end
+
+    def connect_backends!(req, headers, backends)
+      @backends = select_backends(req, headers).select {|b| backends[b]}
+      @backends.unshift master_backend unless @backends.include? master_backend
+      info "Backends for #{req[:method]} #{req[:url]} -> #{@backends}"
+
+      @backends.each do |name|
+        s = server name, backends[name]
+        s.comm_inactivity_timeout = 10
+      end
+    end
+
+    def select_backends(request, headers)
+      callback(:on_select_backends, request, headers) { [master_backend] }
+    end
+
+    def handle(server)
+      self.comm_inactivity_timeout = server.client_timeout
+      self.master_backend = server.master
+
+      @session_id = "%016x" % Random.rand(2**64)
+      info "New connection"
+
+      @callbacks = server.callbacks
+
+      @responses = {}
+      @request = {}
+      @requests = []
+
+      @state = :request
+
+      @parser = HTTP::Parser.new
+      @parser.on_message_begin = proc do
+        @start_time ||= Time.now
+        @state = :request
+      end
+
+      @parser.on_headers_complete = proc do |headers|
+        @request = {
+          :method => @parser.http_method,
+          :path => @parser.request_path,
+          :url => @parser.request_url,
+          :headers => headers
+        }
+        @requests.push @request
+        info "#{@request[:method]} #{@request[:url]}"
+
+        # decide backends on the first request
+        unless @backends
+          connect_backends!(@request, headers, server.backends)
+        end
+
+        if @backends.size > 1
+          info "Multiple backends for this session: Force close connection (disable keep-alives)"
+          headers['Connection'] = 'close'
+        end
+
+        @servers.keys.each do |backend|
+          callback :on_munge_headers, backend, headers
+          relay_to_servers [build_headers(@parser, headers), [backend]]
+        end
+      end
+
+      @parser.on_body = proc do |chunk|
+        relay_to_servers chunk
+      end
+
+      @parser.on_message_complete = proc do
+        @state = :response
+      end
+
+      on_data do |data|
+        begin
+          @parser << data
+        rescue HTTP::Parser::Error
+          info "HTTP parser error: Bad Request"
+          EM.next_tick { close_connection_after_writing }
+        end
+        nil
+      end
+
+      # modify / process response stream
+      on_response do |backend, resp|
+        @responses[backend] ||= {}
+        @responses[backend][:elapsed] = Time.now.to_f - @start_time.to_f
+        @responses[backend][:data] ||= ''
+        @responses[backend][:data] += resp
+
+        resp if backend == master_backend
+      end
+
+      # termination logic
+      on_finish do |backend|
+        # terminate connection (in duplex mode, you can terminate when prod is done)
+        if all_servers_finished?
+          if @backends.all? {|b| @responses[b]}
+            callback :on_backends_finished, @backends, @requests, @responses if @backends.size > 1
+          else
+            info "Server(s) disconnected before response returned: #{@backends.reject {|b| @responses[b]}}"
+          end
+          cleanup!
+        end
+
+        if backend == master_backend
+          info "Master backend closed connection. Closing downstream"
+          :close
+        end
+      end
+    rescue Exception => e
+      info "#{e} - #{e.backtrace}"
+    end
+
+    def info(msg)
+      puts "#{Time.now.strftime('%H:%M:%S')} [#{@session_id}] #{msg}"
+    end
+  end
+end
+