We use `totalMetrics` with a `startTimeInNanos` and a call to `postIngest` to track the 'total time' that ingest is running for a given document. You can imagine it's a little stopwatch that we start and stop. Unfortunately, I think we're stopping the stopwatch just a little too soon, on the start side, we include the time spent transforming the document into an `IngestDocument`, but we do not include the (non-negligible) time transforming from an `IngestDocument` after the pipeline logic has run.

That is, the call to `sourceAsMap` is included:

https://github.com/elastic/elasticsearch/blob/d03664f84f15dcfcb900cf2d43121ca9d89bdda8/server/src/main/java/org/elasticsearch/ingest/IngestService.java#L905

But the call to `source` is not:

https://github.com/elastic/elasticsearch/blob/d03664f84f15dcfcb900cf2d43121ca9d89bdda8/server/src/main/java/org/elasticsearch/ingest/IngestService.java#L948

The consequence of this is that the `total` ingest stats doesn't actually reflect the true time spent in ingest code:

```
GET _nodes/stats?filter_path=nodes.*.ingest
{
  "nodes" : {
    "Aysm6ztIRne2ES5ktyuqtw" : {
      "ingest" : {
        "total" : {
          "count" : 1,
          "time_in_millis" : 110, <--- this should be a little higher than it is
```
server/src/main/java/org/elasticsearch/ingest/IngestService.java
listener.onResponse(AcknowledgedResponse.TRUE);
            return;
        }
        nodeInfoListener.accept(ActionListener.wrap(nodeInfos -> {
            validatePipelineRequest(request, nodeInfos);
            clusterService.submitStateUpdateTask(
                "put-pipeline-" + request.getId(),
                new PutPipelineClusterStateUpdateTask(listener, request),
                ClusterStateTaskConfig.build(Priority.NORMAL, request.masterNodeTimeout()),
                PIPELINE_TASK_EXECUTOR
            );
        }, listener::onFailure));
    }
    public void validatePipelineRequest(PutPipelineRequest request, NodesInfoResponse nodeInfos) throws Exception {
        final Map<String, Object> config = XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
        Map<DiscoveryNode, IngestInfo> ingestInfos = new HashMap<>();
        for (NodeInfo nodeInfo : nodeInfos.getNodes()) {
            ingestInfos.put(nodeInfo.getNode(), nodeInfo.getInfo(IngestInfo.class));
        }
        validatePipeline(ingestInfos, request.getId(), config);
    }
    public static boolean isNoOpPipelineUpdate(ClusterState state, PutPipelineRequest request) {
        IngestMetadata currentIngestMetadata = state.metadata().custom(IngestMetadata.TYPE);
        if (request.getVersion() == null
            && currentIngestMetadata != null
            && currentIngestMetadata.getPipelines().containsKey(request.getId())) {
            var pipelineConfig = XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
            var currentPipeline = currentIngestMetadata.getPipelines().get(request.getId());
            if (currentPipeline.getConfigAsMap().equals(pipelineConfig)) {