如何通过Fluentd将日志字符串解析为OpenSearch可搜索JSON字段？

当前你用ClusterFilter仅完成了日志前缀移除，但JSON内容仍以字符串形式存储，无法直接搜索嵌套字段。以下是两种原生且高效的解决方案，支持嵌套JSON结构解析：

方案1：使用OpenSearch Ingest Pipeline（原生推荐） #

无需额外组件，直接在OpenSearch集群内完成日志结构化处理：

1. 创建JSON解析Pipeline #

PUT _ingest/pipeline/json_log_parser
{
  "description": "移除日志前缀并解析嵌套JSON",
  "processors": [
    // 若ClusterFilter已处理前缀，可跳过此grok步骤，直接处理清理后的message字段
    {
      "grok": {
        "field": "message",
        "patterns": ["%{DATA:log_prefix} %{GREEDYDATA:raw_json}"],
        "ignore_missing": true
      }
    },
    // 核心：将JSON字符串解析为嵌套对象
    {
      "json": {
        "field": "raw_json", // 替换为你清理后的日志字段名（比如直接用message）
        "target_field": "structured_log", // 解析后的嵌套对象存放字段
        "ignore_failure": false // 解析失败时标记日志，方便排查
      }
    },
    // 可选：清理临时字段
    {
      "remove": {
        "field": ["log_prefix", "raw_json"]
      }
    }
  ]
}

2. 关联Pipeline到日志索引 #

在索引模板中设置默认Pipeline，确保所有新写入的日志自动被解析：

PUT _index_template/app_logs_template
{
  "index_patterns": ["app-logs-*"],
  "template": {
    "settings": {
      "index.default_pipeline": "json_log_parser"
    },
    "mappings": {
      "properties": {
        "structured_log": {
          "type": "object" // 声明为object类型，支持嵌套字段搜索
        }
      }
    }
  }
}

方案2：用Logstash处理（若已用Logstash收日志） #

如果日志通过Logstash导入OpenSearch，直接在Logstash配置中添加JSON解析逻辑：

input {
  beats {
    port => 5044 // 示例：通过Filebeat收日志
  }
}

filter {
  # 移除日志前缀（替代ClusterFilter）
  grok {
    match => { "message" => "%{DATA:log_prefix} %{GREEDYDATA:raw_json}" }
    remove_field => ["log_prefix"]
  }
  # 解析嵌套JSON
  json {
    source => "raw_json"
    target => "structured_log"
    remove_field => ["raw_json"]
  }
}

output {
  opensearch {
    hosts => ["http://your-opensearch:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"
  }
}

验证效果 #

在OpenSearch Dashboards的Discover中选择对应索引，structured_log字段会展开为嵌套子字段（比如structured_log.user.id、structured_log.action），可直接用于搜索、聚合和可视化。

常见问题排查 #

• 解析后字段未显示：检查索引映射是否设为object类型，确认Pipeline已正确关联到索引。
• 嵌套JSON解析失败：检查原始日志的JSON格式是否合法（引号、逗号是否匹配），可在Pipeline中添加fail处理器捕获错误日志。

内容的提问来源于stack exchange，提问作者Kaio H. Cunha