BUG: metrics "opendal_operation_duration_seconds_count" gets HTTP request count for chunk reads

[dentiny]

Feature Description

Hi team, I notice for simple 1-1 mapping read + write operations (i.e., read an object from bucket A and write it to bucket B), read and write request rate shows completely different results (read RPS is several times larger than write RPS), both are backed by metrics opendal_operation_duration_seconds_count.

If I don't read it wrong:

• For read operations, the metrics is actually counting for HTTP requests for chunk reader
  • On read operations, we create multiple readers, with each of them containing the full layers:

    opendal/core/core/src/types/read/buffer_stream.rs

    Line 115 in 69114d2

    if let Some(r) = self.generator.next_reader().await? {

• For write operations, things are different, all writes through the same writer created here

  opendal/core/core/src/types/context/write.rs

  Line 113 in 69114d2

  let (_, w) = ctx.acc.write(ctx.path(), ctx.args().clone()).await?;

• In one word, for read operations, the operation count metrics is measuring the HTTP request count, while for write operations, it's measuring the write operation/function call count.

It's not ideal from my perspective:

• Read and write operation counts are completely different if I call Operation::read and Operation::write the same time, which causes user confusion
• Read operation count overlaps with HTTP request count

Problem and Solution

I made my workaround to collect and emit these metrics at application level.
But would be good to calculate these two metrics in the same way for both operations.

Additional Context

No response

Are you willing to contribute to the development of this feature?

• Yes, I am willing to contribute to the development of this feature.

[dentiny]

OK I found the read and write path are handled completely differently.

Write path

Operator::write("path", data)
  → WriteContext::create()
    → ctx.acc.write(path, args)          ← called ONCE, goes through all layers
      → MetricsAccessor::write()         ← creates ONE MetricsWrapper<Writer>
        → inner.write()
          → S3Backend::write()           ← returns MultipartWriter<S3Writer>
    → returns oio::Writer (wrapped by MetricsWrapper)
  → WriteGenerator holds this single writer
  → WriteGenerator::write(chunk1)        ← calls w.write(chunk1) on the SAME writer
    → MetricsWrapper::write()            ← accumulates bytes
      → MultipartWriter::write()
        → S3Writer::write_part()         ← HTTP PUT (internal, below layers)
  → WriteGenerator::write(chunk2)        ← same writer, same MetricsWrapper
    → MetricsWrapper::write()            ← accumulates more bytes
      → MultipartWriter::write()
        → S3Writer::write_part()         ← another HTTP PUT (internal)
  → WriteGenerator::close()
    → MetricsWrapper::close()
      → MultipartWriter::close()
        → S3Writer::complete_part()      ← HTTP POST (internal)
  → MetricsWrapper drops → emits ONE aggregated operation metric

Read path

Operator::reader("path").read(..)
  → Reader::new(ReadContext { acc, path, args, options })
  → BufferStream → ReadGenerator::next_reader()
    → ctx.acc.read(path, args_for_chunk_1)    ← through ALL layers
      → MetricsAccessor::read()               ← creates MetricsWrapper #1
        → S3Backend::read() → HTTP GET
  → ReadGenerator::next_reader()
    → ctx.acc.read(path, args_for_chunk_2)    ← through ALL layers AGAIN
      → MetricsAccessor::read()               ← creates MetricsWrapper #2
        → S3Backend::read() → HTTP GET
  → MetricsWrapper #1 drops → emits metric for chunk 1
  → MetricsWrapper #2 drops → emits metric for chunk 2