Add fluent-ctl command

It sends a simple command to a fluentd process.
The main purpose of this command is that providing a kill command
alternative on Windows. On UNIX, it just only send a UNIX signal to a
specified process.

Usage: fluentdctl COMMAND PID_OR_SIGNAME

Commands:
  shutdown
  restart
  flush
  reload

SIGNAME can be specified by fluentd's "-x" or "--signame" option.
Note that this command doesn't support sending events to windows
service.

Signed-off-by: Takuro Ashie <[email protected]>

bin/fluent-ctl

@@ -0,0 +1,7 @@
+#!/usr/bin/env ruby
+
+here = File.dirname(__FILE__)
+$LOAD_PATH << File.expand_path(File.join(here, '..', 'lib'))
+require 'fluent/command/ctl'
+
+Fluent::Ctl.new.call

lib/fluent/command/ctl.rb

@@ -0,0 +1,132 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'optparse'
+require 'fluent/env'
+require 'fluent/version'
+require 'win32/event' if Fluent.windows?
+
+module Fluent
+  class Ctl
+    DEFAULT_OPTIONS = {}
+    UNIX_SIGNAL_MAP = {
+      shutdown: :TERM,
+      restart: :HUP,
+      flush: :USR1,
+      reload: :USR2,
+    }
+    WINDOWS_EVENT_MAP = {
+      shutdown: "",
+      restart: "HUP",
+      flush: "USR1",
+      reload: "USR2",
+    }
+    if Fluent.windows?
+      COMMAND_MAP = WINDOWS_EVENT_MAP
+    else
+      COMMAND_MAP = UNIX_SIGNAL_MAP
+    end
+
+    def initialize(argv = ARGV)
+      @argv = argv
+      @options = {}
+      @opt_parser = OptionParser.new
+      configure_option_parser
+      @options.merge!(DEFAULT_OPTIONS)
+      parse_options!
+    end
+
+    def help_text
+      text = "\n"
+      if Fluent.windows?
+        text << "Usage: #{$PROGRAM_NAME} COMMAND PID_OR_SIGNAME\n"
+      else
+        text << "Usage: #{$PROGRAM_NAME} COMMAND PID\n"
+      end
+      text << "\n"
+      text << "Commands: \n"
+      COMMAND_MAP.each do |key, value|
+        text << "  #{key}\n"
+      end
+      text
+    end
+
+    def usage(msg = nil)
+      puts help_text
+      if msg
+        puts
+        puts "Error: #{msg}"
+      end
+      exit 1
+    end
+
+    def call
+      if Fluent.windows?
+        call_windows_event(@command, @pid_or_signame)
+      else
+        call_signal(@command, @pid_or_signame)
+      end
+    end
+
+    private
+
+    def call_signal(command, pid)
+      signal = COMMAND_MAP[command.to_sym]
+      Process.kill(signal, pid.to_i)
+    end
+
+    def call_windows_event(command, pid_or_signame)
+      if pid_or_signame =~ /^[0-9]+$/
+        prefix = "fluentd_#{pid_or_signame}"
+      else
+        prefix = pid_or_signame
+      end
+      event_name = COMMAND_MAP[command.to_sym]
+      suffix = event_name.empty? ? "" : "_#{event_name}"
+
+      begin
+        event = Win32::Event.open("#{prefix}#{suffix}")
+        event.set
+        event.close
+      rescue Errno::ENOENT => e
+        puts "Error: Cannot find the fluentd process with the event name: \"#{prefix}\""
+      end
+    end
+
+    def configure_option_parser
+      @opt_parser.banner = help_text
+      @opt_parser.version = Fluent::VERSION
+    end
+
+    def parse_options!
+      @opt_parser.parse!(@argv)
+
+      @command = @argv[0]
+      @pid_or_signame = @argv[1]
+
+      usage("Command isn't specified!") if @command.nil? || @command.empty?
+      usage("Unknown command: #{@command}") unless COMMAND_MAP.has_key?(@command.to_sym)
+
+      if Fluent.windows?
+        usage("PID or SIGNAME isn't specified!") if @pid_or_signame.nil? || @pid_or_signame.empty?
+      else
+        usage("PID isn't specified!") if @pid_or_signame.nil? || @pid_or_signame.empty?
+        usage("Invalid PID: #{pid}") unless @pid_or_signame =~ /^[0-9]+$/
+      end
+    end
+  end
+end
+

test/command/test_ctl.rb

@@ -0,0 +1,57 @@
+require_relative '../helper'
+
+require 'test-unit'
+require 'win32/event' if Fluent.windows?
+
+require 'fluent/command/ctl'
+
+class TestFluentdCtl < ::Test::Unit::TestCase
+  def assert_win32_event(event_name, command, pid_or_svcname)
+    command, event_suffix = data
+    event = Win32::Event.new(event_name)
+    ipc = Win32::Ipc.new(event.handle)
+    ret = Win32::Ipc::TIMEOUT
+
+    wait_thread = Thread.new do
+      ret = ipc.wait(1)
+    end
+    Fluent::Ctl.new([command, pid_or_svcname]).call
+    wait_thread.join
+    assert_equal(Win32::Ipc::SIGNALED, ret)
+  end
+
+  data("shutdown" => ["shutdown", "TERM", ""],
+       "restart" => ["restart", "HUP", "HUP"],
+       "flush" => ["flush", "USR1", "USR1"],
+       "reload" => ["reload", "USR2", "USR2"])
+  def test_commands(data)
+    command, signal, event_suffix = data
+
+    if Fluent.windows?
+      event_name = "fluentd_54321"
+      event_name << "_#{event_suffix}" unless event_suffix.empty?
+      assert_win32_event(event_name, command, "54321")
+    else
+      got_signal = false
+      Signal.trap(signal) do
+        got_signal = true
+      end
+      Fluent::Ctl.new([command, "#{$$}"]).call
+      assert_true(got_signal)
+    end
+  end
+
+  data("shutdown" => ["shutdown", ""],
+       "restart" => ["restart", "HUP"],
+       "flush" => ["flush", "USR1"],
+       "reload" => ["reload", "USR2"])
+  def test_commands_with_winsvcname(data)
+    omit "Only for Windows" unless Fluent.windows?
+
+    command, event_suffix = data
+    event_name = "testfluentdwinsvc"
+    event_name << "_#{event_suffix}" unless event_suffix.empty?
+
+    assert_win32_event(event_name, command, "testfluentdwinsvc")
+  end
+end