#
# Copyright 2020- LIN LI
#
# 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 'fluent/plugin/filter'
require 'json'
require 'sentry-ruby'
require 'audit_log_parser'

module Fluent
  module Plugin
    class LightCoreFilter < Fluent::Plugin::Filter
      Fluent::Plugin.register_filter('light_core', self)

      # 通知相关配置
      config_param :sentry,     :bool,   default: false
      config_param :sentry_dsn, :string, default: ''

      # Application 异常 settings 
      config_param :app_stream,  :string, default: 'stderr'
      config_param :app_message, :array,  default: [],      value_type: :regexp
      config_param :app_status,  :array,  default: ['500'], value_type: :string
      config_param :app_elapsed, :float,  default: 2000

      # Nginx 异常 settings 
      config_param :lb_stream,  :string, default: 'stderr'
      config_param :lb_code,    :array,  default: ['400', '500'], value_type: :string
      config_param :lb_elapsed, :float,  default: 3
      config_param :lb_ignore,  :array,  default: [], value_type: :string

      # MongoDB 异常 settings 
      config_param :mongo_severity,  :array, default: ['F', 'E'], value_type: :string
      config_param :mongo_querytime, :float, default: 100
      config_param :mongo_dataBytes, :float, default: 1073741824
      config_param :mongo_reslen,    :float, default: 20971520

      # 初始化 Sentry
      def start
        super

        if @sentry
          log.info('init sentry')
          Sentry.init do |config|
            config.dsn = @sentry_dsn
          
            # To activate performance monitoring, set one of these options.
            # We recommend adjusting the value in production:
            config.traces_sample_rate = 1

            # config.background_worker_threads = 2
            config.transport.timeout = 10
            config.transport.open_timeout = 10
          end
        end
      end

      # 清理
      def shutdown
        super
      end

      # 主处理
      def filter(tag, time, record)
        
        # 应用
        if ['app', 'service'].include? tag
          record = filter_app(tag, time, record)
          return unless record
          notice('app', record)
        end

        # 负载均衡
        if ['lb'].include? tag
          record = filter_lb(tag, time, record)
          return unless record
          notice('lb', record)
        end

        # 数据库
        if ['master', 'secondary', 'arbiter', 'third'].include? tag
          record = filter_mongo(tag, time, record)
          return unless record
          notice('mongo', record)
        end

        if ['syslog.messages', 'syslog.secure', 'syslog.audit'].include? tag
          record = filter_syslog(tag, time, record)
          return unless record
        end

        record['environment'] = ENV['FLUENTD_ENV']
        record['node'] = ENV['NODE_IP']

        # 其他
        record

      end

      # Parse syslog
      def filter_syslog(tag, time, record)

        if (tag == 'syslog.audit')
          line = record['message']
          return record unless line

          record = AuditLogParser.parse_line(line, flatten: false)
          record['time'] = Time.at(record["header"]["msg"][/[0-9]+/].to_i).to_s
          return record
        end

        record['time'] = Time.at(time).to_s
        return record
      end

      # Parse the application log
      def filter_app(tag, time, record)
        file = record['file'].split('/').last.split('_')           # Parse log file name
        log = record['log']                                        # Get detailed log content

        # Set common items
        record['cid'] = file[0]                                    # container id
        record['cname'] = file[0].split('-')[1]                    # container name
        record['ctime'] = record['time']                           # container time

        # Delete useless content
        record.delete('log')
        record.delete('file')
        record.delete('time')
        
        # Standard AP log [2020-07-06T09:21:51.121] [A] [INFO] xxxxx
        if /^\[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{3}\] \[A\]/.match(log)

          item = /^\[(?<time>[^\]]*)\] \[(?<component>[^\]]*)\] \[(?<levle>[^\]]*)\] (?<line>[^\ ]*) - - (?<uid>[^\ ]*) - (?<message>.*)$/.match(log)

          record['time'] = item[:time]
          record['component'] = item[:component]
          record['levle'] = item[:levle]
          record['line'] = item[:line]
          record['uid'] = item[:uid]
          record['message'] = item[:message]

          return record
        end

        # Standard Access log [2020-07-06T09:21:51.121] [I] [GET] xxxxx
        if /^\[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{3}\] \[I\]/.match(log)

          item = /^\[(?<time>[^\]]*)\] \[(?<component>[^\]]*)\] \[(?<method>[^\]]*)\] (?<url>[^\ ]*) (?<status>[^\ ]*) (?<size>[^\ ]*) (?<uid>[^\ ]*) (?<elapsed>[^\ ]*) (?<addr>[^\ ]*)$/.match(log)

          record['time'] = item[:time]
          record['component'] = item[:component]
          record['method'] = item[:method]
          record['url'] = item[:url]
          record['status'] = item[:status]
          record['size'] = str_to_num(item[:size])
          record['uid'] = item[:uid]
          record['elapsed'] = str_to_num(item[:elapsed])
          record['addr'] = item[:addr].gsub(/\n$/, '')

          return record
        end

        # Console log
        record['message'] = log.gsub(/\n$/, '')
        record
      end

      # https://docs.nginx.com/nginx/admin-guide/monitoring/logging/
      def filter_lb(tag, time, record)

        file = record['file'].split('/').last.split('_')
        log = record['log']

        record['cid'] = file[0]
        record['cname'] = tag
        record['ctime'] = record['time']

        record.delete('log')
        record.delete('file')
        record.delete('time')
        
        # access log
        if /^[^ ]+ [^ ]+ [^ ]+ \[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+\d{2}:\d{2}\]/.match(log)
          item = /^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)" "(?<elapsed>[^\"]*)" "(?<requestid>[^\"]*)" "(?<sessionid>[^\"]*)")?/.match(log)

          record['remote'] = item[:remote]
          record['host'] = item[:host]
          record['user'] = item[:user]
          record['time'] = item[:time]
          record['method'] = item[:method]
          record['path'] = item[:path]
          record['code'] = item[:code]
          record['size'] = str_to_num(item[:size])
          record['referer'] = item[:referer]
          record['agent'] = item[:agent]
          record['elapsed'] = str_to_num(item[:elapsed])
          record['requestid'] = item[:requestid]
          record['sessionid'] = item[:sessionid]

          return record
        end

        # error log
        if /^\d{4}\/\d{2}\/\d{2} \d{2}:\d{2}:\d{2} \[error\]/.match(log)
          item = /^(?<time>[^ ]+ [^ ]+) \[(?<level>.*)\] (?<pid>\d*)#(?<tid>[^:]*): \*(?<cid>\d*) (?<message>.*)$/.match(log)

          record['time'] = record['ctime']
          record['level'] = item[:level]
          record['message'] = item[:message]
          record['process'] = item[:pid]     # process id
          record['thread'] = item[:tid]      # thread id
          record['counter'] = item[:cid]     # counter

          detail = /request: "(?<method>[^ ]*) (?<path>[^"]*)"/.match(item[:message])
          unless detail.nil?
            record['method'] = detail[:method]
            record['path'] = detail[:path]
          end

          detail = /referrer: "(?<referrer>[^"]*)"/.match(item[:message])
          unless detail.nil?
            record['referrer'] = detail[:referrer]
          end

          return record
        end

        # other log
        record['message'] = log.gsub(/\n$/, '')
        record
      end

      # https://docs.mongodb.com/manual/reference/log-messages/
      def filter_mongo(tag, time, record)

        file = record['file'].split('/').last.split('_')
        logging = record['log']

        record['cid'] = file[0]
        record['cname'] = tag
        record['ctime'] = record['time']

        record.delete('log')
        record.delete('file')
        record.delete('time')

        # 版本4.4开始，默认日志为json格式
        begin
          item = JSON.parse(logging)
        rescue JSON::ParserError
          log.warn('json parse error : ', logging)
          return nil
        end

        record['time'] = item['t']['$date']
        record['severity'] = item['s']
        record['component'] = item['c']
        record['context'] = item['ctx']
        record['identifier'] = item['id']
        record['message'] = item['msg']
        
        attributes = item['attr']
        if attributes
          record['querytime'] = attributes['durationMillis']
          record['collection'] = attributes['ns']
          record['reslen'] = attributes['reslen']
          storage = attributes['storage']
          if storage
            data = storage['data']
            if data
              record['bytesRead'] = data['bytesRead']
              record['bytesWritten'] = data['bytesWritten']
            end
          end
          record['attr'] = attributes
        end

        record['tags'] = item['tags']
        record['truncated'] = item['truncated']
        record['size'] = item['size']
        record
      end

      # 确认是否发送通知
      def notice(tag, record)

        unless @sentry
          return record
        end

        if tag == 'app'

          if @app_message.length > 0 && record['message']
            @app_message.each do |pattern|
              if pattern.match(record['message'])
                send(tag, record['message'], record)
                break
              end
            end
          end

          if @app_status.length > 0 && record['status']
            message = 'Status code abnormal : ' + record['url']
            send(tag, message, record) if @app_status.include?(record['status'])
          end

          if @app_elapsed > 0 && record['elapsed']
            message = 'Slow process : ' + record['url']
            send(tag, message, record) if record['elapsed'].to_f >= @app_elapsed
          end

          return record
        end

        if tag == 'lb'

          if @lb_ignore && record['path'] && @lb_ignore.include?(record['path'])
            return record
          end

          if @lb_stream && record['stream']
            send(tag, 'Stderror', record) if record['stream'] == @lb_stream
          end

          if @lb_code.length > 0 && record['code']
            message = 'Status code abnormal : ' + record['path']
            send(tag, message, record) if @lb_code.include?(record['code'])
          end

          if @lb_elapsed > 0 && record['elapsed']
            message = 'Slow request : ' + record['path']
            send(tag, message, record) if record['elapsed'].to_f >= @lb_elapsed
          end
    
          return record
        end

        if tag == 'mongo'
          if @mongo_severity.length > 0 && record['severity']
            message = 'Severity level abnormal : ' + record['severity']
            send(tag, message, record) if @mongo_severity.include?(record['severity'])
          end

          if @mongo_querytime > 0 && record['querytime']
            message = 'Slow query'
            message = message + ' : ' + record['collection'] if record['collection']
            send(tag, message, record) if record['querytime'].to_f >= @mongo_querytime
          end

          if @mongo_dataBytes > 0 && record['bytesRead']
             message = 'bytesRead data is:' + record['bytesRead'].to_s
             message = message + ' : ' + record['collection'] if record['collection']
             send(tag, message, record) if record['bytesRead'].to_f >= @mongo_dataBytes
          end

          if @mongo_dataBytes > 0 && record['bytesWritten']
            message = 'bytesWritten data is:' + record['bytesWritten'].to_s
            message = message + ' : ' + record['collection'] if record['collection']
            send(tag, message, record) if record['bytesWritten'].to_f >= @mongo_dataBytes
          end

          if @mongo_reslen > 0 && record['reslen']
            message = 'reslen data is:' + record['reslen'].to_s
            message = message + ' : ' + record['collection'] if record['collection']
            send(tag, message, record) if record['reslen'].to_f >= @mongo_reslen
          end

          return record
        end

        record
        
      end

      # 发送UDP请求
      def send(tag, message, record)
        Sentry.capture_message(message, :extra => record, :tags => {'log' => tag})
        log.info('msg has been sent to sentry : ', message)
      end

      # 转数字
      def str_to_num(str)
        return 0 if (str == '-')
        return str.to_i if (str =~ /^\d+$/)
        return str.to_f if (str =~ /^\d+\.\d+$/)
        str
      end

    end
  end
end