OpenSearch Ingest Pipeline无法保留@timestamp的date类型问题

我正尝试通过Filebeat将JSON数据发送至OpenSearch的logpipeline.json处理管道。对部分字段的类型转换操作正常，但遇到@timestamp字段的问题：原本为date类型的@timestamp，在将JSON数据解析到根级对象data后，data.@timestamp显示为string类型，即便未对其做任何转换操作。如何在处理管道完成转换后，仍保留@timestamp字段的date类型？

相关数据与配置 #

原始JSON日志数据 #

{"@timestamp":"2022-11-08T10:07:05+00:00","client":"10.x.x.x","server_name":"example.stack.com","server_port":"80","server_protocol":"HTTP/1.1","method":"POST","request":"/example/api/v1/","request_length":"200","status":"500","bytes_sent":"598","body_bytes_sent":"138","referer":"","user_agent":"Java/1.8.0_191","upstream_addr":"10.x.x.x:10376","upstream_status":"500","gzip_ratio":"","content_type":"application/json","request_time":"6.826","upstream_response_time":"6.826","upstream_connect_time":"0.000","upstream_header_time":"6.826","remote_addr":"10.x.x.x","x_forwarded_for":"10.x.x.x","upstream_cache_status":"","ssl_protocol":"TLSv","ssl_cipher":"xxxx","ssl_session_reused":"r","request_body":"{\"date\":null,\"sourceType\":\"BPM\",\"processId\":\"xxxxx\",\"comment\":\"Process status: xxxxx: \",\"user\":\"xxxx\"}","response_body":"{\"statusCode\":500,\"reasonPhrase\":\"Internal Server Error\",\"errorMessage\":\"xxxx\"}","limit_req_status":"","log_body":"1","connection_upgrade":"close","http_upgrade":"","request_uri":"/example/api/v1/","args":""}

Filebeat输出配置（到OpenSearch） #

# ---------------------------- Elasticsearch Output ----------------------------
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["192.168.29.117:9200"]
  pipeline: logpipeline
  #index: "filebeatelastic-%{[agent.version]}-%{+yyyy.MM.dd}"
  index: "nginx_dev-%{+yyyy.MM.dd}"
  # Protocol - either `http` (default) or `https`.
  protocol: "https"
  ssl.enabled: true
  ssl.verification_mode: none

  # Authentication credentials - either API key or username/password.
  #api_key: "id:api_key"
  username: "filebeat"
  password: "filebeat"

OpenSearch处理管道（logpipeline.json） #

{"description":"Logging Pipeline","processors":[{"json":{"field":"message","target_field":"data"}},{"date":{"field":"data.@timestamp","formats":["ISO8601"]}},{"convert":{"field":"data.body_bytes_sent","type":"integer","ignore_missing":true,"ignore_failure":true}},{"convert":{"field":"data.bytes_sent","type":"integer","ignore_missing":true,"ignore_failure":true}},{"convert":{"field":"data.request_length","type":"integer","ignore_missing":true,"ignore_failure":true}},{"convert":{"field":"data.request_time","type":"float","ignore_missing":true,"ignore_failure":true}},{"convert":{"field":"data.upstream_connect_time","type":"float","ignore_missing":true,"ignore_failure":true}},{"convert":{"field":"data.upstream_header_time","type":"float","ignore_missing":true,"ignore_failure":true}},{"convert":{"field":"data.upstream_response_time","type":"float","ignore_missing":true,"ignore_failure":true}}]}

处理管道模拟结果 #

{"docs":[{"doc":{"_index":"_index","_id":"_id","_source":{"index_date":"2022.11.08","@timestamp":"2022-11-08T12:07:05.000+02:00","message":"\"{ \"@timestamp\": \"2022-11-08T10:07:05+00:00\", \"client\": \"10.x.x.x\", \"server_name\": \"example.stack.com\", \"server_port\": \"80\", \"server_protocol\": \"HTTP/1.1\", \"method\": \"POST\", \"request\": \"/example/api/v1/\", \"request_length\": \"200\", \"status\": \"500\", \"bytes_sent\": \"598\", \"body_bytes_sent\": \"138\", \"referer\": \"\", \"user_agent\": \"Java/1.8.0_191\", \"upstream_addr\": \"10.x.x.x:10376\", \"upstream_status\": \"500\", \"gzip_ratio\": \"\", \"content_type\": \"application/json\", \"request_time\": \"6.826\", \"upstream_response_time\": \"6.826\", \"upstream_connect_time\": \"0.000\", \"upstream_header_time\": \"6.826\", \"remote_addr\": \"10.x.x.x\", \"x_forwarded_for\": \"10.x.x.x\", \"upstream_cache_status\": \"\", \"ssl_protocol\": \"TLSv\", \"ssl_cipher\": \"xxxx\", \"ssl_session_reused\": \"r\", \"request_body\": \"{\\\"date\\\":null,\\\"sourceType\\\":\\\"BPM\\\",\\\"processId\\\":\\\"xxxxx\\\",\\\"comment\\\":\\\"Process status: xxxxx: \\\",\\\"user\\\":\\\"xxxx\\\"}\", \"response_body\": \"{\\\"statusCode\\\":500,\\\"reasonPhrase\\\":\\\"Internal Server Error\\\",\\\"errorMessage\\\":\\\"xxxx\\\"}\", \"limit_req_status\": \"\", \"log_body\": \"1\", \"connection_upgrade\": \"close\", \"http_upgrade\": \"\", \"request_uri\": \"/example/api/v1/\", \"args\": \"\"}\",\"data":{"server_name":"example.stack.com","request":"/example/api/v1/","referer":"","log_body":"1","upstream_addr":"10.x.x.x:10376","body_bytes_sent":138,"upstream_header_time":6.826,"ssl_cipher":"xxxx","response_body":"\"{\"statusCode\":500,\"reasonPhrase\":\"Internal Server Error\",\"errorMessage\":\"xxxx\"}\"","upstream_status":"500","request_time":6.826,"upstream_cache_status":"","content_type":"application/json","client":"10.x.x.x","user_agent":"Java/1.8.0_191","ssl_protocol":"TLSv","limit_req_status":"","remote_addr":"10.x.x.x","method":"POST","gzip_ratio":"","http_upgrade":"","bytes_sent":598,"request_uri":"/example/api/v1/","x_forwarded_for":"10.x.x.x","args":"","@timestamp":"2022-11-08T10:07:05+00:00","upstream_connect_time":0.0,"request_body":"\"{\"date\":null,\"sourceType\":\"BPM\",\"processId\":\"xxxxx\",\"comment\":\"Process status: xxxxx: \",\"user\":\"xxxx\"}\"","request_length":200,"ssl_session_reused":"r","server_port":"80","upstream_response_time":6.826,"connection_upgrade":"close","server_protocol":"HTTP/1.1","status":"500"}}},"_ingest":{"timestamp":"2023-01-18T08:06:35.335066236Z"}}}]}]

问题核心：json处理器解析后data.@timestamp为字符串，当前date处理器仅将解析后的时间写入根级@timestamp（默认行为），未修改data.@timestamp本身的类型。提供两种解决方式：

方式一：修改date处理器，覆盖data.@timestamp类型 #

调整date处理器的target_field参数，明确将解析后的日期写回data.@timestamp，同时可保留根级@timestamp：

{"description":"Logging Pipeline","processors":[
  {"json":{"field":"message","target_field":"data"}},
  {"date":{"field":"data.@timestamp","formats":["ISO8601"],"target_field":"data.@timestamp"}},
  // 可选：同步更新根级@timestamp
  {"date":{"field":"data.@timestamp","formats":["ISO8601"]}},
  {"convert":{"field":"data.body_bytes_sent","type":"integer","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"data.bytes_sent","type":"integer","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"data.request_length","type":"integer","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"data.request_time","type":"float","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"data.upstream_connect_time","type":"float","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"data.upstream_header_time","type":"float","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"data.upstream_response_time","type":"float","ignore_missing":true,"ignore_failure":true}}
]}

方式二：在Filebeat提前解析JSON，减少管道处理 #

在Filebeat输入配置中直接解析JSON到根级，保留原始@timestamp的date类型，管道仅处理字段类型转换：

filebeat.inputs:
- type: log
  paths:
    - /path/to/your/logs/*.log
  json.keys_under_root: true
  json.add_error_key: true
  # 可选：用日志中的@timestamp覆盖Filebeat自动生成的时间
  json.overwrite_keys: true

修改OpenSearch处理管道，移除json处理器：

{"description":"Logging Pipeline","processors":[
  {"convert":{"field":"body_bytes_sent","type":"integer","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"bytes_sent","type":"integer","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"request_length","type":"integer","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"request_time","type":"float","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"upstream_connect_time","type":"float","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"upstream_header_time","type":"float","ignore_missing":true,"ignore_failure":true}},
  {"convert":{"field":"upstream_response_time","type":"float","ignore_missing":true,"ignore_failure":true}}
]}

内容的提问来源于stack exchange，提问作者sb9