Merge pull request #1 from ripienaar/management

add a working agent to manage the emulator

README.md

@@ -1,35 +1,36 @@
 # What?
 
-This is a tool that creates multiple Choria instances in memory in a single process.
+This is a tool that creates multiple Choria instances in memory in a single process for the purpose of load testing a Choria infrastructure wrt middleware and Federation performance.
 
 Each instance can have a number of emulated agents, belong to many sub collectives and generally you'll be able to interact with them from the normal Choria `mco` CLI. Each instance will make a NATS connection just like Choria does and make the same subscriptions. 
 
 On my MBP I can run 2 000 instances of Choria along with a Choria Broker and the Choria client all on the same laptop and response times are around 1.5 seconds for all nodes.
 
-The idea is that you will use many VMs, say a few 100, deploy standard Choria to them along with an agent that will be provided here to manage a big network of emulated Choria instances.
+The idea is that you will use many VMs, say a few 100, deploy standard Choria to them along with an agent, that will be provided here, to manage a big network of emulated Choria instances.
 
-Each of these 100s of VMs can run lets say a thousand Choria instances at a time and you can point them at different topologies of NATS, Federation etc and do tests with different concurrencies and payload sizes.
+Each of these 100s of VMs can run, lets say, a thousand Choria instances at a time and you can point them at different topologies of NATS, Federation etc and do tests with different concurrencies and payload sizes.
 
 ```
-usage: choria-emulator --name=NAME --instances=INSTANCES --agents=AGENTS [<flags>]
+usage: choria-emulator --name=NAME --instances=INSTANCES [<flags>]
 
 Emulator for Choria Networks
 
 Flags:
  --help Show context-sensitive help (also try --help-long and --help-man).
  --version Show application version.
- --name=NAME Instance name
+ --name=NAME Instance name prefix
  -i, --instances=INSTANCES Number of instances to start
  -a, --agents=1 Number of emulated agents to start
  --collectives=1 Number of emulated subcollectives to create
  -c, --config=CONFIG Choria configuration file
  --tls Enable TLS on the NATS connections
  --verify Enable TLS certificate verifications on the NATS connections
  --server=SERVER ... NATS Server pool, specify multiple times (eg one:4222)
+ -p, --http-port=8080 Port to listen for /debug/vars
 ```
 
 ## Agents
-When you specify the creation of 10 agents you will get a series of agents called `emulated0`, `emulated1` and so forth.
+When you specify the creation of 10 agents you will get a series of agents called `emulated0`, `emulated1` and so forth. The main motivation for this is to discover the limits of NATS since each agent involves multiple queues.
 
 These agents are all identical and have just one action `generate` that takes a `size` argument. It will create a reply message with a string reply equal in size to what was requested.
 
@@ -38,7 +39,7 @@ This is good test the impact of varying sizes of payload on your infrastructure.
 Additionally it has the standard `discovery` agent so `mco ping` and so forth works. Filters wise it only support the `agent` filter which is sufficient for this kind of testing.
 
 ```
-$ mco rpc emulated0 generate size=100 -I test-1 -j
+$ mco rpc emulated0 generate size=10 -I test-1 -j
 [
  {
  "agent": "emulated0",
@@ -47,7 +48,7 @@ $ mco rpc emulated0 generate size=100 -I test-1 -j
  "statuscode": 0,
  "statusmsg": "OK",
  "data": {
- "message": "0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789"
+ "message": "0123456789"
  }
  }
 ]
@@ -56,4 +57,8 @@ $ mco rpc emulated0 generate size=100 -I test-1 -j
 ## Subcollectives
 Subcollectives are supported, by default it belongs to the typical `mcollective` sub collective.
 
-If you ask if to belong to more than 3 using the `--collectives` option it will subscribe to collectives `mcollective`, `collective1` and `collective2`.
+If you ask if to belong to 3 using the `--collectives` option it will subscribe to collectives `mcollective`, `collective1` and `collective2`.
+
+The motivation for allowing multiples collectives is that each collective multiplies the amount of middleware subscriptions.
+
+With 10 agents in one collective it would make 11 subscriptions, in 2 collectives it would make 22 and so forth.

agent/emulated0.ddl

@@ -0,0 +1,22 @@
+metadata :name => "emulated0",
+ :description => "Choria Agent emulated by choria-emulator",
+ :author => "R.I.Pienaar <rip@devco.net>",
+ :license => "Apache-2.0",
+ :version => "0.0.1",
+ :url => "http://choria.io",
+ :timeout => 120
+
+requires :mcollective => "2.9.0"
+
+action "generate", :description => "Generates random data of a given size" do
+ input :size,
+ :prompt => "Size",
+ :description => "Amount of text to generate",
+ :type => :integer,
+ :optional => true,
+ :default => 20
+
+ output :message,
+ :description => "Generated Message",
+ :display_as => "Message"
+end

agent/emulator.ddl

@@ -0,0 +1,151 @@
+metadata :name => "emulator",
+ :description => "choria-emulator manager agent",
+ :author => "R.I.Pienaar <rip@devco.net>",
+ :license => "Apache-2.0",
+ :version => "0.0.1",
+ :url => "http://choria.io",
+ :timeout => 60
+
+requires :mcollective => "2.9.0"
+
+action "status", :description => "Status of the running emulator" do
+ display :always
+
+ input :port,
+ :description => "Port to query for status",
+ :prompt => "Monitor Port",
+ :type => :integer,
+ :optional => false,
+ :default => 8080
+
+ output :name,
+ :description => "Instance name",
+ :display_as => "Name",
+ :default => "unknown"
+
+ output :running,
+ :description => "Is the instance running",
+ :display_as => "Running",
+ :default => false
+
+ output :pid,
+ :description => "Running PID",
+ :display_as => "PID",
+ :default => -1
+
+ output :tls,
+ :description => "TLS Enabled",
+ :display_as => "TLS",
+ :default => false
+
+ output :memory,
+ :description => "Memory used in bytes",
+ :display_as => "Memory (B)",
+ :default => 0
+
+ summarize do
+ aggregate summary(:running)
+ aggregate summary(:tls)
+ end
+end
+
+action "download", :description => "Downloads the emulator binary" do
+ input :http,
+ :description => "HTTP or HTTPS URL to fetch the file from",
+ :prompt => "Emulator source URL",
+ :type => :string,
+ :optional => false,
+ :maxlength => "256",
+ :validation => "."
+
+ output :success,
+ :description => "true if the emulator was downloaded",
+ :display_as => "Downloaded"
+
+ output :size,
+ :description => "Size of file downloaded",
+ :display_as => "Size"
+
+ summarize do
+ aggregate summary(:success)
+ aggregate summary(:size)
+ end
+end
+
+action "stop", :description => "Stops the running emulator instance" do
+ input :port,
+ :description => "Port to query for status",
+ :prompt => "Monitor Port",
+ :type => :integer,
+ :optional => false,
+ :default => 8080
+
+ output :status,
+ :description => "true if the emulator stopped",
+ :display_as => "Stopped"
+
+ summarize do
+ aggregate summary(:status)
+ end
+end
+
+action "start", :description => "Start an emulator instance" do
+ input :name,
+ :prompt => "Name",
+ :description => "Instance Name",
+ :type => :string,
+ :optional => true,
+ :maxlength => "16",
+ :validation => '^\w+$'
+
+ input :instances,
+ :prompt => "Instances",
+ :description => "Number of simulated choria instances the emulator will host",
+ :type => :integer,
+ :optional => false,
+ :default => 1
+
+ input :agents,
+ :prompt => "Agents",
+ :description => "Number of emulated* agents the emulator will host",
+ :type => :integer,
+ :optional => false,
+ :default => 1
+
+ input :collectives,
+ :prompt => "Subcollectives",
+ :description => "Number of subcollective the emulator will join",
+ :type => :integer,
+ :optional => false,
+ :default => 1
+
+ input :servers,
+ :prompt => "Servers to connect to",
+ :description => "Comma separated list of host:port pairs",
+ :type => :string,
+ :maxlength => "256",
+ :optional => true,
+ :validation => "."
+
+ input :monitor,
+ :description => "Port to listen for monitoring requests",
+ :prompt => "Monitor Port",
+ :type => :integer,
+ :optional => false,
+ :default => 8080
+
+ input :tls,
+ :description => "Run with TLS enabled",
+ :prompt => "TLS",
+ :type => :boolean,
+ :optional => false,
+ :default => false
+
+ output :status,
+ :description => "true if the emulator started",
+ :display_as => "Started"
+
+ summarize do
+ aggregate summary(:status)
+ end
+end

agent/emulator.rb

@@ -0,0 +1,138 @@
+require 'net/http'
+
+module MCollective
+ module Agent
+ class Emulator<RPC::Agent
+ action "download" do
+ reply[:success] = false
+
+ FileUtils.mkdir_p("/tmp/choria-emulator")
+
+ if request[:http]
+ begin
+ download_http(request[:http], "/tmp/choria-emulator/choria-emulator")
+ rescue
+ reply.fail!("Failed to download %s: %s: %s" % [request[:http], $!.class, $!.to_s])
+ end
+
+ FileUtils.chmod(0755, "/tmp/choria-emulator/choria-emulator")
+
+ stat = File::Stat.new("/tmp/choria-emulator/choria-emulator")
+ reply[:size] = stat.size
+ reply[:success] = true
+ else
+ reply.fail("No valid download location given")
+ end
+ end
+
+ action "start" do
+ unless File.executable?("/tmp/choria-emulator/choria-emulator")
+ reply.fail!("Cannot start /tmp/choria-emulator/choria-emulator does not exist or is not executable")
+ end
+
+ if up?(request[:monitor])
+ reply.fail!("Cannot start, emulator is already running")
+ end
+
+ args = []
+
+ args << "--name %s" % request[:name]
+ args << "--instances %d" % request[:instances]
+ args << "--http-port %d" % request[:monitor]
+ args << "--config /dev/null"
+
+ args << "--agents %d" % request[:agents] if request[:agents]
+ args << "--collectives %d" % request[:collectives] if request[:collectives]
+ args << "--server %s" % request[:servers].gsub(" ", "") if request[:servers]
+ args << "--tls" if request[:tls]
+
+ out = []
+ err = []
+ Log.info("Running: %s" % args.join(" "))
+
+ run('(/tmp/choria-emulator/choria-emulator %s 2>&1 >> /tmp/choria-emulator/log &) &' % args.join(" "), :stdout => out, :stderr => err)
+
+ sleep 1
+
+ reply[:status] = up?(request[:monitor])
+ end
+
+ action "status" do
+ if down?(request[:port])
+ reply[:running] = false
+ break
+ end
+
+ status = emulator_status(request[:port])
+
+ reply[:name] = status["name"]
+ reply[:running] = true
+ reply[:pid] = status["config"]["pid"]
+ reply[:tls] = status["config"]["TLS"] == 1
+ reply[:memory] = status["memstats"]["Sys"]
+ end
+
+ action "stop" do
+ reply[:status] = false
+
+ if up?(request[:port])
+ pid = emulator_pid(request[:port])
+
+ reply.fail!("Could not determine PID for running emulator") unless pid
+
+ Process.kill("HUP", pid)
+ sleep 1
+ reply[:status] = down?(request[:port])
+ end
+ end
+
+ def up?(port)
+ Log.debug(emulator_status(port).inspect)
+ emulator_status(port)["status"] == "up"
+ rescue
+ Log.warn("%s: %s" % [$!.class, $!.to_s])
+ false
+ end
+
+ def down?(port)
+ !up?(port)
+ end
+
+ def emulator_pid(port)
+ emulator_status(port)["config"]["pid"]
+ end
+
+ def emulator_status(port=8080)
+ uri = URI.parse("http://localhost:%d/debug/vars" % port)
+ response = Net::HTTP.get_response(uri)
+
+ out = {
+ "status" => "up",
+ "code" => response.code
+ }
+
+ if response.code == "200"
+ Log.debug(response.body)
+ out.merge!(JSON.parse(response.body))
+ end
+
+ out
+ end
+
+ def download_http(url, target)
+ uri = URI.parse(url)
+ out = File.open(target, "wb")
+
+ begin
+ Net::HTTP.get_response(uri) do |resp|
+ resp.read_body do |segment|
+ out.write(segment)
+ end
+ end
+ ensure
+ out.close
+ end
+ end
+ end
+ end
+end