如何从插入中生成弹性的新文档

Question

目前我使用Jaeger监控我的应用程序，作为后端，它使用弹性搜索，我希望能够根据一些标准将这些信息聚合到新的文档中，以便生成一些报告。

从插入具有以下标签"phase=end"的文档开始：

{
"_index": "jaeger-span-2022-03-30",
"_type": "_doc",
"_id": "7fso238BngwX41T6Cr9y",
"_version": 1,
"_score": null,
"fields": {
"traceID": [
"0cefe26bed7464436c43519e7fcbf6c2"
],
"duration": [
1898679
],
"spanID": [
"74b463687cfaf503"
],
"startTimeMillis": [
"2022-03-30T14:08:011.642Z"
],
"references": [
{
"spanID": [
"8657c748a0508e8b"
],
"traceID": [
"0cefe26bed7464436c43519e7fcbf6c2"
],
"refType": [
"CHILD_OF"
]
}
],
"process.serviceName": [
"a09-002"
],
"startTime": [
1648649289644801
],
"operationName": [
"br.com.flow.items.FinalOperation.execute"
],
"tags": [
{
"type": [
"string"
],
"value": [
"end"
],
"key": [
"phase"
]
}
]
}
}

在跟踪开始时创建的文档如下所示：

{
"_index": "jaeger-span-2022-03-30",
"_type": "_doc",
"_id": "7fso238BngwX41T6Cr9y",
"_version": 1,
"_score": null,
"fields": {
"traceID": [
"0cefe26bed7464436c43519e7fcbf6c2"
],
"duration": [
1898679
],
"spanID": [
"74b463687cfaf503"
],
"startTimeMillis": [
"2022-03-30T14:08:09.642Z"
],
"references": [
{
"spanID": [
"8657c748a0508e8b"
],
"traceID": [
"0cefe26bed7464436c43519e7fcbf6c2"
],
"refType": [
"CHILD_OF"
]
}
],
"process.serviceName": [
"a09-002"
],
"startTime": [
1648649289642801
],
"operationName": [
"br.com.flow.items.InitialOperation.execute"
]
}
}

我想将数据连接起来，形成一个新的文档，如下所示：

{
"fields": {
"traceID": [
"0cefe26bed7464436c43519e7fcbf6c2"
],
"duration": [
2000
],
"startTime": [
1648649289642801
],
"endTime": [
1648649289644801
],
"process.serviceName": [
"a09-002"
]
}
}

为此，我需要做以下步骤：

1-观察带有上述标签的文件插入事件

2-计算endTime，它基本上是包含阶段的文档的startTime+持续时间=结束标签

3-计算持续时间，为此需要捕获第一个插入的文档的startTime，该文档与带有标签的文档具有相同的traceID，然后从中减去之前计算的endTime值。

如何执行这些任务并生成新文档？最初我想使用CDC来完成这个过程，但弹性搜索似乎不支持这个功能。

Answer 1

我最终走了另一条路，我将跟踪发送到Kafka，并制作了一个流来分析数据，然后将合并结果保存在数据库中。