Class: Fluent::MysqlBinlogFlydataInput

Inherits:
MysqlBinlogInput
  • Object
show all
Includes:
MysqlBinlogFlydataInputPreference
Defined in:
lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb

Constant Summary

Constants included from MysqlBinlogFlydataInputPreference

Fluent::MysqlBinlogFlydataInputPreference::CUSTOM_CONFIG_PARAMS

Instance Method Summary collapse

Methods included from MysqlBinlogFlydataInputPreference

included

Constructor Details

#initialize ⇒ MysqlBinlogFlydataInput 



37
38
39
40
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 37

def initialize
  super
  install_custom_signal_handler
end

Instance Method Details

#configure(conf) ⇒ Object 



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 45

def configure(conf)
  super
  unless File.exists?(@position_file)
    raise "No position file(#{@position_file}). Initial synchronization is required before starting."
  end
  load_custom_conf
  $log.info "mysql host:\"#{@host}\" username:\"#{@username}\" database:\"#{@database}\" tables:\"#{@tables}\""
  @tables = @tables.split(/,\s*/)
  sync_fm = Flydata::FileUtil::SyncFileManager.new(nil) # Passing nil for data_entry as this class does not use methods which require data_entry

  @context = Mysql::Context.new(
    database: @database, tables: @tables,
    tag: @tag, sync_fm: sync_fm
  )
  @record_dispatcher = Mysql::FlydataBinlogRecordDispatcher.new(@context)
end

#event_listener(event) ⇒ Object 



111
112
113
114
115
116
117
118
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 111

def event_listener(event)
  @record_dispatcher.dispatch(event)
rescue Exception => e
  position = File.open(@position_file) {|f| f.read }
  $log.error "error occured while processing #{event.event_type} event at #{position}\n#{e.message}\n#{$!.backtrace.join("\n")}"
  # Not reraising a StandardError because the underlying code can't handle an error well.
  raise unless e.kind_of?(StandardError)
end

#install_custom_signal_handler ⇒ Object

Hack: All that has been added here is Fluent::Engine.shutdown_source. This should be in fluentd's supervisor#install_main_process_signal_handlers



144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 144

def install_custom_signal_handler
    trap :USR1 do
      $log.debug "fluentd main process get SIGUSR1"
      $log.info "force flushing buffered events"
      #@log.reopen!

      # Creating new thread due to mutex can't lock
      # in main thread during trap context
      Thread.new {
        begin 
          Fluent::Engine.shutdown_source
          Fluent::Engine.flush!
          $log.debug "flushing thread: flushed"
        rescue Exception => e
          $log.warn "flushing thread error: #{e}"
        end
      }.run
    end
end

#run ⇒ Object 



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 82

def run
  start_kodama(mysql_url) do |c|
    c.binlog_position_file = @position_file
    c.connection_retry_limit = @retry_limit
    c.connection_retry_wait = @retry_wait
    c.log_level = @log_level.to_sym
    @listen_events.each do |event_type|
      $log.trace { "registered binlog event listener '#{event_type}'" }
      c.send("on_#{event_type}", &method(:event_listener))
    end
  end
rescue
  $log.error "unexpected error. exception: #{$!.class.to_s}, error: #{$!.to_s}\n#{$!.backtrace.join("\n")}"
  raise
rescue SignalException
  $log.debug "signal exception. exception: #{$!.class.to_s}, error: #{$!.to_s}"
  raise
rescue Exception
  $log.error "unexpected fatal error. exception: #{$!.class.to_s}, error: #{$!.to_s}\n#{$!.backtrace.join("\n")}"
  raise
end

#shutdown ⇒ Object 



120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 120

def shutdown
  if @thread and @thread.alive?
    $log.info "Requesting stop Kodama"
    @kodama_client.stop_request
    if wait_till_safe_to_stop
      $log.info "Killing Kodama client"
      Thread.kill(@thread)
    else
      $log.error "Unable to stop Kodama"
    end
  end
end

#start ⇒ Object 



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 62

def start
  super
  positions_path = @context.sync_fm.table_positions_dir_path
  Dir.mkdir positions_path unless File.exists? positions_path
rescue Binlog::Error
  if (/basic_string::_M_replace_aux/ === $!.to_s)
    # TODO Fix the root cause in mysql-replication-listener
    $log.error "a mysql-replication-listener error.  This could have been caused by one of the following reasons.\n- Failed on connect: Your host is blocked because of many connection errors; unblock with 'mysqladmin flush-hosts'\n"
  else
    $log.error "unexpected mysql-replication-listener error. exception: #{$!.class.to_s}, error: #{$!.to_s}\n#{$!.backtrace.join("\n")}"
  end
  raise
rescue Exception
  $log.error "unexpected fatal error. exception: #{$!.class.to_s}, error: #{$!.to_s}\n#{$!.backtrace.join("\n")}"
  raise
end

#start_kodama(options, &block) ⇒ Object 



104
105
106
107
108
109
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 104

def start_kodama(options, &block)
  @kodama_client = Kodama::Client.new(Kodama::Client.mysql_url(options))
  @kodama_client.logger = $log
  block.call(@kodama_client)
  @kodama_client.start
end

#wait_till_safe_to_stop ⇒ Object 



133
134
135
136
137
138
139
140
 
# File 'lib/flydata/fluent-plugins/in_mysql_binlog_flydata.rb', line 133

def wait_till_safe_to_stop
  retry_count = 5
  1.upto(retry_count) do |i|
    return true if @kodama_client.safe_to_stop?
    sleep 3
  end
  false
end