Logstash Ruby 过滤器深度解析：释放数据处理的无限可能

Logstash 是一个强大的数据处理管道，能够从多个数据源摄取数据，对其进行转换，然后发送到指定的目标。过滤器插件在这个过程中起着关键作用，它们对通过管道的数据执行特定操作。

虽然 Logstash 内置了多个过滤器来处理常见任务（如解析、丰富和修改数据），但有时您会遇到需要超越标准过滤器功能的自定义逻辑场景。这就是 Ruby 过滤器插件 发挥作用的地方。

Ruby 过滤器插件允许您直接在 Logstash 管道中执行自定义 Ruby 代码。当标准过滤器不够用时，Ruby 过滤器使您能够处理复杂的数据转换、实现自定义业务逻辑或与外部系统集成。

作为 Elastic 的咨询架构师，我经常看到客户在使用 Logstash 进行数据处理时，面对标准过滤器在复杂数据操作或自定义逻辑方面的限制而苦恼。在这些情况下，Ruby 过滤器可以帮助克服这些挑战。

Ruby 过滤器的典型使用场景包括：

• 深度嵌套数据操作：修改复杂的 JSON 结构、数组中的数组，或根据内容动态重构数据

• 高级字符串处理：从非结构化文本中解析和提取结构化数据

• 实现复杂业务逻辑：创建需要条件逻辑、循环或复杂计算的自定义转换

创建 Logstash 管道时，应将配置文件放在 etc/logstash/conf.d 目录中。或者，您可以使用 -f 选项在手动启动 Logstash 时指定配置文件的路径，这样可以轻松试验您的管道。

配置文件应具有 .conf 扩展名。

要使用 Ruby 过滤器，请在 Logstash 管道配置文件的过滤器部分定义一个 ruby 过滤器。以下是基本示例：

filter {
  ruby {
    code => "event.set('new_field', 'Hello from Ruby!')"
  }
}

这个内联 Ruby 过滤器在您的 Logstash 配置中定义了一个 Ruby 过滤器实例。code 参数提供了 Logstash 将为此过滤器处理的每个事件执行的内联 Ruby 脚本。

在该脚本内部，有一个可用的 event 变量，它代表事件本身。Event 对象包含发送到 Logstash 的原始数据以及在 Logstash 过滤阶段创建的任何其他字段。您可以通过 Logstash Event API（如 event.get() 和 event.set()）访问这些字段。

在此示例代码中，event.set('new_field', 'Hello from Ruby!') 将名为 new_field 的新字段设置为字符串值 Hello from Ruby!。您可以根据需要在此 code 块中添加任何其他代码。

注意：此 event 对象不是常规的 Ruby 哈希对象，尽管它充当键值类型数据容器。查看官方文档以了解更多关于 Event API 的信息。

对于简单的转换，内联 Ruby 代码很方便。但是，对于复杂逻辑或可重用函数，建议将代码移动到外部 Ruby 脚本中。这提高了可维护性并保持 Logstash 管道配置的整洁。

首先，创建一个 Ruby 脚本并将其保存为 my_ruby_script.rb。脚本必须定义一个处理事件的 filter 方法。它以事件对象作为参数，该对象代表当前正在处理的事件。filter 方法需要返回要发出的事件数组。要删除事件，请返回空数组。

例如，以下脚本读取 message 字段，计算其长度，并将结果存储在名为 message_length 的新字段中：

# my_ruby_script.rb
def filter(event)
  message = event.get('message')
  if message
    event.set('message_length', message.length)
  end
  return [event]
end

filter {
  ruby {
    path => "/path/to/my_ruby_script.rb"
  }
}

使用外部脚本时，请确保文件存在并具有正确的权限。

现在，每个事件都传递给 my_ruby_script.rb 中的 filter 方法并由其处理。

这种方法帮助您更有效地管理复杂逻辑，使测试、调试和重用 Ruby 代码变得更容易。

Logstash 事件是 Logstash 处理的核心数据结构。它可以包含各种字段，包括嵌套数据结构，如数组和哈希。Ruby 过滤器允许您轻松操作这些嵌套结构。

Ruby 过滤器可以处理嵌套数据结构，如哈希和数组，允许您修改或添加这些结构内的字段。这在处理复杂数据格式（如 JSON）时很有用。

filter {
  ruby {
    code => '
      user_data = event.get("user")
      if user_data && user_data.is_a?(Hash)
        user_data["full_name"] = "#{user_data["first_name"]} #{user_data["last_name"]}"
        event.set("user", user_data)
      end
    '
  }
}

Ruby 过滤器也可用于将单个事件拆分为多个事件。这在您有包含项目数组的单个事件并且想要为每个项目创建单独事件时很有用。

注意：无论是 Elasticsearch 的摄取管道还是 Beats/Elastic Agent 的处理器都不支持拆分事件。这是 Logstash 最强的用例之一。

使用 split 过滤器

您可以使用 split 过滤器根据指定字段将事件拆分为多个事件。但是，如果您需要在拆分期间执行其他转换或逻辑，可以将 Ruby 过滤器与 split 过滤器结合使用。

在以下示例中，我们有一个 RSS 源作为单行 XML 文本。它包含多个 <item> 元素。Ruby 过滤器用于从 XML 中提取 <item> 元素并将它们存储在名为 items 的新字段中。然后使用 split 过滤器根据 items 字段将事件拆分为多个事件。

input {
  stdin { }
}


filter {
  ruby {
    code => '
      message = event.get("message")
      items = message.scan(/<item>.*?<\/item>/m)
      event.set("items", items)
    '
  }
  
  split {
    field => "items"
  }
  
  mutate {
    remove_field => ["message", "items"]
  }
}


output {
  stdout { codec => rubydebug }
}

使用内联 Ruby 脚本

您也可以使用内联 Ruby 脚本通过使用 event.clone 方法和 new_event_block 变量（如 new_event_block.call(new_event)）将单个事件拆分为多个事件。这允许您基于原始事件创建新事件，同时保留其数据。

以下是如何使用 Ruby 过滤器将单个事件拆分为多个事件的示例：

filter {
  ruby {
    code => '
      items = event.get("items")
      if items && items.is_a?(Array)
        items.each do |item|
          new_event = event.clone
          new_event.set("item", item)
          new_event_block.call(new_event)
        end
        event.cancel
      end
    '
  }
}

使用外部 Ruby 脚本

您也可以使用外部 Ruby 脚本将单个事件拆分为多个事件。

配置文件：

filter {
  ruby {
    path => "/path/to/split_event.rb"
  }
}

def filter(event)
  items = event.get("items")
  events = []
  
  if items && items.is_a?(Array)
    items.each do |item|
      new_event = event.clone
      new_event.set("item", item)
      new_event.remove("items")
      events << new_event
    end
  else
    events << event
  end
  
  return events
end

Logstash exec 输入插件允许您执行外部命令，其输出将成为 Logstash 的事件。命令的输出将存储在事件的 message 字段中。

通常，系统命令的输出是人类可读的，但不是结构化的 JSON 或 Logstash 可以轻松解析的其他格式。为了处理这个问题，您可以使用 Ruby 过滤器来解析输出并从中提取信息。

以下是使用 exec 输入插件执行 ps -ef 命令的示例，该命令列出类 Unix 系统上所有正在运行的进程。输出将由 Ruby 过滤器处理，以提取有关每个进程的相关信息。

input {
  exec {
    command => "ps -ef"
    interval => 60
  }
}


filter {
  ruby {
    code => '
      message = event.get("message")
      lines = message.split("\n")
      
      # Skip header line
      if lines.length > 1
        lines[1..-1].each do |line|
          # Parse ps -ef output
          if match = line.match(/^\s*(\S+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(.+)$/)
            new_event = event.clone
            new_event.set("uid", match[1])
            new_event.set("pid", match[2].to_i)
            new_event.set("ppid", match[3].to_i)
            new_event.set("cpu_usage", match[4].to_i)
            new_event.set("stime", match[5])
            new_event.set("tty", match[6])
            new_event.set("time", match[7])
            new_event.set("command", match[8])
            new_event_block.call(new_event)
          end
        end
        event.cancel
      end
    '
  }
}

Ruby 过滤器插件允许您使用内置的 Ruby 库，这对于各种任务都非常有用。例如，您可以使用 json 库解析 JSON 字符串或使用 date 库操作日期。

以下是使用 json 库解析存储在字段中的 JSON 字符串的示例：

filter {
  ruby {
    code => '
      require "json"
      json_string = event.get("json_data")
      if json_string
        begin
          parsed_data = JSON.parse(json_string)
          event.set("parsed_json", parsed_data)
        rescue JSON::ParserError => e
          event.set("json_parse_error", e.message)
        end
      end
    '
  }
}

为了避免每次都需要库，您应该外部化您的 Ruby 代码，以便可以在 Ruby 过滤器脚本的开头使用 require 语句。这将加载库一次并使其可用于脚本中。

要检查环境中可用的库，可以通过在 Ruby 过滤器中运行以下代码来列出内置库：

filter {
  ruby {
    code => 'puts $LOAD_PATH'
  }
}

def filter(event)
  begin
    # 您的处理逻辑
    message = event.get('message')
    # 处理逻辑...
  rescue => e
    event.set('ruby_error', e.message)
    event.tag('_rubyexception')
  end
  
  return [event]
end

使用日志输出：在 Ruby 代码中添加 puts 语句进行调试

事件标记：使用 event.tag() 方法标记处理过的事件

逐步测试：从简单逻辑开始，逐步增加复杂性

Logstash Ruby 过滤器允许您自定义和扩展 Logstash 管道的功能。通过本文，我们覆盖了使用 Ruby 过滤器的基础知识并提供了高级用法示例。

通过利用 Ruby 过滤器，您可以处理需要自定义逻辑或高级操作的复杂数据处理任务。无论您是在处理嵌套数据结构、拆分事件，还是解析和转换复杂/非结构化文本为结构化 JSON，Ruby 过滤器都提供了满足您特定需求的灵活性。

Ruby 过滤器是 Logstash 工具箱中的一个强大工具，当标准过滤器无法满足需求时，它能够释放数据处理的无限可能。掌握 Ruby 过滤器，让您的数据管道更加强大和灵活！

感谢您关注新智锦绣科技（北京）有限公司！作为 Elastic 的 Elite 合作伙伴及 EnterpriseDB 在国内的唯一代理和服务合作伙伴，我们始终致力于技术创新和优质服务，帮助企业客户实现数据平台的高效构建与智能化管理。无论您是关注 Elastic 生态系统，还是需要 EnterpriseDB 的支持，我们都将为您提供专业的技术支持和量身定制的解决方案。

欢迎关注我们，获取更多技术资讯和数字化转型方案，共创美好未来！