Addition of internal consumer metrics

[jmacd]

Adds three OTel metric instruments with instrumentation on:

• records consumed
• schema resets
• memory in use.

Note that I have not added tests, which would delay us a lot for little benefit. There are existing debug-level assertions that cover the memory accounting. To test this experimentally, I used a recorded data file and a configuration like:

receivers:
  file/input:
    path: "${env:DIR}/data.json.zstd"
    format: json
    compression: zstd
    throttle: 0
  otelarrow/loopback:
    protocols:
      grpc:
        endpoint: 127.0.0.1:8082
        max_recv_msg_size_mib: 24
        
exporters:
  otelarrow/forward:
    endpoint: 127.0.0.1:8082
    wait_for_ready: true
    arrow:
      disabled: false
      num_streams: 1
      disable_downgrade: true
    tls:
      insecure: true
    retry_on_failure:
      enabled: false
    sending_queue:
      enabled: false
      num_consumers: 1
    timeout: 30s
  logging:
  file/output:
    path: "${env:DIR}/output.json.zstd"
    format: json
    compression: zstd

extensions:
  pprof:

processors:
  batch:
    send_batch_size: 1000
    send_batch_max_size: 1100
    timeout: 10s

service:
  extensions: [pprof]
  pipelines:
    traces/first:
      receivers: [file/input]
      processors: [batch]
      exporters: [otelarrow/forward]
    traces/second:
      receivers: [otelarrow/loopback]
      processors: []
      exporters: [logging, file/output]
      
  telemetry:
    resource:
      "service.name": "data-replayer"
    metrics:
      address: 127.0.0.1:8889
      level: detailed
    logs:
      level: debug

Then I see metrics (via Prometheus) at :8889/metrics, like:

# HELP otelcol_arrow_batch_records_total 
# TYPE otelcol_arrow_batch_records_total counter
otelcol_arrow_batch_records_total{stream_unique="4797068c"} 52445
# HELP otelcol_arrow_memory_inuse 
# TYPE otelcol_arrow_memory_inuse gauge
otelcol_arrow_memory_inuse{stream_unique="4797068c",where="library"} 5.6499008e+07
otelcol_arrow_memory_inuse{stream_unique="4797068c",where="user"} 0
# HELP otelcol_arrow_schema_resets_total 
# TYPE otelcol_arrow_schema_resets_total counter
otelcol_arrow_schema_resets_total{payload_type="RESOURCE_ATTRS",stream_unique="4797068c"} 1
otelcol_arrow_schema_resets_total{payload_type="SPANS",stream_unique="4797068c"} 7
otelcol_arrow_schema_resets_total{payload_type="SPAN_ATTRS",stream_unique="4797068c"} 4
otelcol_arrow_schema_resets_total{payload_type="SPAN_EVENTS",stream_unique="4797068c"} 3
otelcol_arrow_schema_resets_total{payload_type="SPAN_EVENT_ATTRS",stream_unique="4797068c"} 5
otelcol_arrow_schema_resets_total{payload_type="SPAN_LINKS",stream_unique="4797068c"} 3
otelcol_arrow_schema_resets_total{payload_type="SPAN_LINK_ATTRS",stream_unique="4797068c"} 1

Note that memory inuse appears to rise slowly.

[jmacd]

(By the way, acknowledging that Prometheus will not be a good way to monitor metrics that have a unique attribute created every few minutes. This instrumentation maybe too expensive to run in practice, depending on vendor support for delta temporality and SDK support for "forgetting", but I would encourage us to over-instrument now and reduce later.

[moh-osman3]

basic question: what is the purpose of nesting these calls? i.e. what is the reason for returning placeholder{} and what is the difference between this and function signatures that allow sequential defers like

defer c.unuseChangeObserve()
defer c.inuseChangeByCaller()

[jmacd]

See https://go.dev/doc/effective_go#defer for some detail.

The arguments to the deferred function (which include the receiver if the function is a method) are evaluated when the defer executes, not when the call executes.

This idiom allows me to use 1 line. The equivalent would be:

c.inuseChangeByCaller()
defer c.inuseChangeObserve()

if after learning about defer's evaluation order you think that's more readable, I'll change it!

[moh-osman3]

is this function thread safe? i.e. will setting c.lastInuseValue here have a risk of overwriting/being overwritten or are all consumers isolated?

[jmacd]

The consumers are isolated. During an earlier discussion @lquerel asked me to avoid using asynchronous instruments, which would have required additional concurrency, so presently we have no atomic variables and no concurrent use. That's why we're wrangling a synchronous instrument to do what is ordinarily done through an observer.

[lquerel]

LGTM

[lquerel]

Not strictly equivalent to the previous code. With this modification we have a single allocator for all the payloads instead of one allocator per payload type. It's probably ok but that could explain why we observe a different behavior with the previous approach.