parse: add protobuf parser and native histograms

specs/expfmt/SPEC.md

@@ -79,25 +79,29 @@ pub trait Parser<'a> {
     /// Advances to the next entry in the exposition stream, returning `Ok(None)` at EOF.
     fn next_entry(&mut self) -> Result<Option<Entry>, ParseError>;
 
-    fn help(&self) -> (&'a [u8], Cow<'a, [u8]>);
-    fn metric_type(&self) -> (&'a [u8], MetricType);
-    fn unit(&self) -> (&'a [u8], &'a [u8]);
+    // These slices borrow from the parser, because the protobuf parser builds them.
+    fn help(&self) -> (&[u8], Cow<'_, [u8]>);
+    fn metric_type(&self) -> (&[u8], MetricType);
+    fn unit(&self) -> (&[u8], &[u8]);
     fn comment(&self) -> &'a [u8];
 
     /// Returns the raw series bytes, optional sample timestamp in ms, and float64 value.
-    fn series(&self) -> (&'a [u8], Option<i64>, f64);
+    fn series(&self) -> (&[u8], Option<i64>, f64);
 
     /// Returns the raw series bytes, optional timestamp in ms, and parsed native histogram.
-    fn histogram(&self) -> (&'a [u8], Option<i64>, HistogramRef<'a>);
+    fn histogram(&self) -> (&[u8], Option<i64>, HistogramRef<'_>);
 
-    /// Returns the start timestamp in ms if present via `st@...` or `_created`.
-    fn start_timestamp(&self) -> Option<i64>;
+    /// Returns the start timestamp in ms if present via `st@...`, `_created`, or the protobuf
+    /// `start_timestamp` field.
+    fn start_timestamp(&mut self) -> Option<i64>;
 
     /// Populates the caller's reusable label buffer for the current sample.
     fn labels(&self, buf: &mut LabelsBuffer<'a>);
 
-    /// Iterates over exemplars attached to the current sample.
-    fn exemplars(&self, buf: &mut ExemplarsBuffer<'a>) -> Result<(), ParseError>;
+    /// Fills `buf` with the labels of the next exemplar of the current sample and returns its
+    /// value and optional timestamp in ms. A native histogram can have several exemplars, so
+    /// call it until it returns `None`.
+    fn exemplar(&mut self, buf: &mut LabelsBuffer<'a>) -> Option<(f64, Option<i64>)>;
 }
 ```
 
@@ -179,7 +183,9 @@ To verify that `ArrowDecoder` meets the [`open-telemetry/otel-arrow#3277`](https
 
 ### Phase 3: OpenMetrics 2.0 and Prometheus protobuf decoding
 - Implement Layer 0, Layer 1, and Layer 2 support for OpenMetrics 2.0 matching [`openmetrics2parse.go`](https://github.com/prometheus/prometheus/blob/main/model/textparse/openmetrics2parse.go) (native histograms in text format, `st@` start timestamps, composite values) and add `specs/expfmt/openmetrics_2_0.abnf`.
-- Implement delimited and text protobuf decoding (`FmtProtoDelim`, `FmtProtoText`, `FmtProtoCompact`), including native sparse histograms and NHCB conversion matching [`protobufparse.go`](https://github.com/prometheus/prometheus/blob/main/model/textparse/protobufparse.go) and [`nhcbparse.go`](https://github.com/prometheus/prometheus/blob/main/model/textparse/nhcbparse.go).
+- Implement delimited protobuf decoding (`FmtProtoDelim`), including native sparse histograms, start timestamps, and exemplars, matching [`protobufparse.go`](https://github.com/prometheus/prometheus/blob/main/model/textparse/protobufparse.go).
+- Histograms are only returned as native histograms, converting explicit-bucket histograms to NHCB (#37). Classic `_bucket`, `_sum`, and `_count` histogram series are left out on purpose; we can add classic support later if users need it ([discussion](https://github.com/prometheus/scrape.rs/pull/25#discussion_r4476081239)).
+- Out of scope: `FmtProtoText` and `FmtProtoCompact`, because Prometheus only scrapes `encoding=delimited` ([config.go#L584](https://github.com/prometheus/prometheus/blob/v0.315.1/config/config.go#L584)).
 - Extend the Go `expfmt` encoder harness to cover `FmtOpenMetrics_2_0_0` and `FmtProtoDelim`.
 
 ### Phase 4: Relabeling hooks, pooling, benchmarks, fuzzing, and initial release

src/arrow.rs

@@ -94,7 +94,7 @@ impl<'input> ActiveParser<'input> {
     }
 
     #[inline]
-    fn metric_type(&self) -> (&'input [u8], MetricType) {
+    fn metric_type(&self) -> (&[u8], MetricType) {
         match self {
             Self::Prom(p) => p.metric_type(),
             Self::OpenMetrics(p) => p.metric_type(),
@@ -103,7 +103,7 @@ impl<'input> ActiveParser<'input> {
     }
 
     #[inline]
-    fn unit(&self) -> (&'input [u8], &'input [u8]) {
+    fn unit(&self) -> (&[u8], &[u8]) {
         match self {
             Self::Prom(p) => p.unit(),
             Self::OpenMetrics(p) => p.unit(),
@@ -453,7 +453,8 @@ impl ArrowDecoder {
             };
 
             match entry {
-                Entry::Comment => {}
+                // `Entry::Histogram` arrives with #22.
+                Entry::Comment | Entry::Histogram => {}
                 Entry::Help => self.ingest_help_entry(parser, opts, default_ts_ns)?,
                 Entry::Type => self.ingest_type_entry(parser, opts, default_ts_ns)?,
                 Entry::Unit => self.ingest_unit_entry(parser, opts, default_ts_ns)?,

src/arrow/options.rs

@@ -156,6 +156,10 @@ pub enum DecodeError {
     #[error("invalid metric type {0:?}")]
     InvalidMetricType(String),
 
+    /// Malformed protobuf payload.
+    #[error("invalid protobuf: {0}")]
+    InvalidProtobuf(String),
+
     /// Invalid UTF-8 sequence in a metric name, help string, label name, or label value.
     #[error("invalid UTF-8 at byte {offset}")]
     InvalidUtf8 {
@@ -261,6 +265,7 @@ impl DecodeError {
             ParseError::MissingEof => Self::MissingEof,
             ParseError::DataAfterEof { offset } => Self::DataAfterEof { offset },
             ParseError::InvalidUnit { metric, unit } => Self::InvalidUnit { metric, unit },
+            ParseError::InvalidProtobuf(message) => Self::InvalidProtobuf(message),
         }
     }
 }

src/parse.rs

@@ -23,12 +23,16 @@ mod omlex;
 mod omparse;
 mod promlex;
 mod promparse;
+#[cfg(feature = "dto")]
+mod protobufparse;
 
 pub(crate) use float::parse_go_float;
 pub use omparse::OpenMetricsParser;
 pub(crate) use omparse::{OmScratch, OpenMetricsParserState};
 pub use promparse::PromParser;
 pub(crate) use promparse::{ByteSpan, LabelSpan, PromParserState, unescape_bytes_into};
+#[cfg(feature = "dto")]
+pub use protobufparse::ProtobufParser;
 
 /// Label name under which [`Parser::labels`] stores the metric name.
 pub const METRIC_NAME_LABEL: &str = "__name__";
@@ -53,8 +57,81 @@ pub enum Entry {
     Comment,
     /// A sample with a float64 value. Read it with [`Parser::series`] and [`Parser::labels`].
     Series,
+    /// A sample with a native histogram value. Read it with [`Parser::histogram`] and
+    /// [`Parser::labels`].
+    Histogram,
+}
+
+/// Bucket span of a native histogram, Go's `histogram.Span`.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct Span {
+    /// Gap to the previous span, or the start index of the first span.
+    pub offset: i32,
+    /// Number of consecutive buckets.
+    pub length: u32,
+}
+
+/// Counter reset hint of a native histogram, Go's `histogram.CounterResetHint`.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+#[non_exhaustive]
+#[repr(u8)]
+pub enum CounterResetHint {
+    /// No hint, Go's `UnknownCounterReset`.
+    #[default]
+    Unknown = 0,
+    /// The histogram is a gauge, Go's `GaugeType`.
+    Gauge = 3,
+}
+
+/// Native histogram. With `C = u64` and `B = i64` it is Go's `histogram.Histogram`, with
+/// delta-encoded bucket counts. With `C = B = f64` it is Go's `histogram.FloatHistogram`,
+/// with absolute bucket counts.
+#[derive(Clone, Debug, Default, PartialEq)]
+pub struct Histogram<C, B> {
+    /// Counter reset hint.
+    pub counter_reset_hint: CounterResetHint,
+    /// Resolution of the exponential buckets.
+    pub schema: i32,
+    /// Width of the zero bucket.
+    pub zero_threshold: f64,
+    /// Observations in the zero bucket.
+    pub zero_count: C,
+    /// Total number of observations.
+    pub count: C,
+    /// Sum of observations.
+    pub sum: f64,
+    /// Spans of the positive buckets.
+    pub positive_spans: Vec<Span>,
+    /// Positive bucket counts.
+    pub positive_buckets: Vec<B>,
+    /// Spans of the negative buckets.
+    pub negative_spans: Vec<Span>,
+    /// Negative bucket counts.
+    pub negative_buckets: Vec<B>,
+}
+
+/// Native histogram value returned by [`Parser::histogram`].
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum HistogramRef<'h> {
+    /// Integer counts, Go's `*histogram.Histogram`.
+    Int(&'h Histogram<u64, i64>),
+    /// Float counts, Go's `*histogram.FloatHistogram`.
+    Float(&'h Histogram<f64, f64>),
 }
 
+static EMPTY_HISTOGRAM: Histogram<u64, i64> = Histogram {
+    counter_reset_hint: CounterResetHint::Unknown,
+    schema: 0,
+    zero_threshold: 0.0,
+    zero_count: 0,
+    count: 0,
+    sum: 0.0,
+    positive_spans: Vec::new(),
+    positive_buckets: Vec::new(),
+    negative_spans: Vec::new(),
+    negative_buckets: Vec::new(),
+};
+
 /// Metric type declared by a `# TYPE` line.
 ///
 /// Mirrors `model.MetricType` in `prometheus/common/model`.
@@ -123,6 +200,10 @@ pub enum ParseError {
     #[error("invalid sample value {0:?}")]
     InvalidValue(String),
 
+    /// A protobuf payload is malformed or holds a value Go's protobuf parser rejects.
+    #[error("invalid protobuf: {0}")]
+    InvalidProtobuf(String),
+
     /// A sample timestamp is not a valid int64.
     #[error("invalid timestamp {0:?}")]
     InvalidTimestamp(String),
@@ -220,6 +301,9 @@ impl<'a> LabelsBuffer<'a> {
 /// Streaming parser over an exposition payload, mirroring Go's `textparse.Parser`.
 ///
 /// Accessors return unspecified values when called for a different [`Entry`] kind.
+///
+/// Byte slices returned by the help, type, unit, series, and histogram accessors borrow from
+/// the parser, because the protobuf parser builds them. Text parsers return slices of the input.
 pub trait Parser<'a> {
     /// Advances to the next entry in the exposition stream, returning `Ok(None)` at EOF.
     fn next_entry(&mut self) -> Result<Option<Entry>, ParseError>;
@@ -228,21 +312,27 @@ pub trait Parser<'a> {
     fn pos(&self) -> usize;
 
     /// Returns the metric name and unescaped help text of the current [`Entry::Help`].
-    fn help(&self) -> (&'a [u8], Cow<'a, [u8]>);
+    fn help(&self) -> (&[u8], Cow<'_, [u8]>);
 
     /// Returns the metric name and type of the current [`Entry::Type`].
-    fn metric_type(&self) -> (&'a [u8], MetricType);
+    fn metric_type(&self) -> (&[u8], MetricType);
 
     /// Returns the text of the current [`Entry::Comment`], including the leading `#`.
     fn comment(&self) -> &'a [u8];
 
     /// Returns the metric name and unit of the current [`Entry::Unit`].
-    fn unit(&self) -> (&'a [u8], &'a [u8]) {
+    fn unit(&self) -> (&[u8], &[u8]) {
         (b"", b"")
     }
 
     /// Returns the raw series bytes, optional sample timestamp in ms, and float64 value.
-    fn series(&self) -> (&'a [u8], Option<i64>, f64);
+    fn series(&self) -> (&[u8], Option<i64>, f64);
+
+    /// Returns the raw series bytes, optional sample timestamp in ms, and native histogram of
+    /// the current [`Entry::Histogram`].
+    fn histogram(&self) -> (&[u8], Option<i64>, HistogramRef<'_>) {
+        (b"", None, HistogramRef::Int(&EMPTY_HISTOGRAM))
+    }
 
     /// Clears `buf` and fills it with the current sample's labels, sorted by name, matching Go:
     /// the metric name is stored under [`METRIC_NAME_LABEL`], and `le` on histograms and
@@ -251,15 +341,17 @@ pub trait Parser<'a> {
 
     /// Clears `buf`, fills it with the exemplar labels of the current sample sorted by name,
     /// and returns the exemplar value and optional timestamp in ms. Returns `None` when the
-    /// sample has no exemplar or it was already returned, like Go.
+    /// sample has no exemplar or it was already returned, like Go. A native histogram can
+    /// have several exemplars, so call it until it returns `None`.
     fn exemplar(&mut self, buf: &mut LabelsBuffer<'a>) -> Option<(f64, Option<i64>)> {
         buf.clear();
         None
     }
 
-    /// Returns the start timestamp in ms of the current sample, read from its `_created` line.
-    /// Takes `&mut self` because, like Go, it caches the last hit and turns on `_created`
-    /// skipping, so later `_created` lines are not returned as series.
+    /// Returns the start timestamp in ms of the current sample, read from its `_created` line
+    /// in OpenMetrics or its `start_timestamp` field in protobuf. Takes `&mut self` because the
+    /// OpenMetrics parser, like Go, caches the last hit and turns on `_created` skipping, so
+    /// later `_created` lines are not returned as series. The protobuf parser only reads the field.
     fn start_timestamp(&mut self) -> Option<i64> {
         None
     }