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
}
            while (processor instanceof WrappingProcessor wrappingProcessor) {
                if (clazz.isAssignableFrom(wrappingProcessor.getInnerProcessor().getClass())) {
                    processors.add(clazz.cast(wrappingProcessor.getInnerProcessor()));
                }
                processor = wrappingProcessor.getInnerProcessor();
                
                
                if (wrappingProcessor == processor) {
                    break;
                }
            }
        }
        return processors;
    }
    public <P extends Processor> Collection<String> getPipelineWithProcessorType(Class<P> clazz, Predicate<P> predicate) {
        List<String> matchedPipelines = new LinkedList<>();
        for (PipelineHolder holder : pipelines.values()) {
            String pipelineId = holder.pipeline.getId();
            List<P> processors = getProcessorsInPipeline(pipelineId, clazz);
            if (processors.isEmpty() == false && processors.stream().anyMatch(predicate)) {
                matchedPipelines.add(pipelineId);
            }
        }
        return matchedPipelines;
    }
    public synchronized void reloadPipeline(String id) throws Exception {
        PipelineHolder holder = pipelines.get(id);
        Pipeline updatedPipeline = Pipeline.create(id, holder.configuration.getConfigAsMap(), processorFactories, scriptService);
        Map<String, PipelineHolder> updatedPipelines = new HashMap<>(this.pipelines);