From b500f999d27309b00377f8e7062c53fa3d1e2223 Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Sun, 3 Nov 2024 15:18:18 +0800
Subject: [PATCH 01/11] 1

---
 metrics-exporter-prometheus/Cargo.toml        |   6 +-
 .../src/exporter/mod.rs                       |   5 +
 .../src/exporter/remote_write.rs              |  69 +++++
 .../src/exporter/remote_write_proto.rs        | 275 ++++++++++++++++++
 4 files changed, 354 insertions(+), 1 deletion(-)
 create mode 100644 metrics-exporter-prometheus/src/exporter/remote_write.rs
 create mode 100644 metrics-exporter-prometheus/src/exporter/remote_write_proto.rs

diff --git a/metrics-exporter-prometheus/Cargo.toml b/metrics-exporter-prometheus/Cargo.toml
index 434e336f..21c23385 100644
--- a/metrics-exporter-prometheus/Cargo.toml
+++ b/metrics-exporter-prometheus/Cargo.toml
@@ -17,11 +17,12 @@ categories = ["development-tools::debugging"]
 keywords = ["metrics", "telemetry", "prometheus"]
 
 [features]
-default = ["http-listener", "push-gateway"]
+default = ["http-listener", "push-gateway","remote-write"]
 async-runtime = ["tokio", "hyper-util/tokio"]
 http-listener = ["async-runtime", "ipnet", "tracing", "_hyper-server"]
 uds-listener = ["http-listener"]
 push-gateway = ["async-runtime", "tracing", "_hyper-client"]
+remote-write = ["_hyper-client","async-runtime","dep:prost","dep:snap","dep:prometheus-parse","dep:prost"]
 _hyper-server = ["http-body-util", "hyper/server", "hyper-util/server-auto"]
 _hyper-client = [
     "http-body-util",
@@ -48,7 +49,10 @@ metrics-util = { version = "^0.18", path = "../metrics-util", default-features =
     "registry",
     "summary",
 ] }
+prometheus-parse = {version = "0.2.5", optional = true}
+prost = {workspace = true, optional = true}
 quanta = { workspace = true }
+snap = { version = "1.1.1", optional = true}
 thiserror = { workspace = true }
 tokio = { workspace = true, optional = true }
 tracing = { workspace = true, optional = true }
diff --git a/metrics-exporter-prometheus/src/exporter/mod.rs b/metrics-exporter-prometheus/src/exporter/mod.rs
index d10c0336..490daa64 100644
--- a/metrics-exporter-prometheus/src/exporter/mod.rs
+++ b/metrics-exporter-prometheus/src/exporter/mod.rs
@@ -71,4 +71,9 @@ mod http_listener;
 #[cfg(feature = "push-gateway")]
 mod push_gateway;
 
+#[cfg(feature = "remote-write")]
+mod remote_write;
+#[cfg(feature = "remote-write")]
+mod remote_write_proto;
+
 pub(crate) mod builder;
diff --git a/metrics-exporter-prometheus/src/exporter/remote_write.rs b/metrics-exporter-prometheus/src/exporter/remote_write.rs
new file mode 100644
index 00000000..a0ce0544
--- /dev/null
+++ b/metrics-exporter-prometheus/src/exporter/remote_write.rs
@@ -0,0 +1,69 @@
+use std::time::Duration;
+
+use http_body_util::{BodyExt, Collected, Full};
+use hyper::{body::Bytes, Method, Request, Uri};
+use hyper_util::{client::legacy::Client, rt::TokioExecutor};
+use tracing::error;
+
+use crate::PrometheusHandle;
+
+use super::ExporterFuture;
+
+// Creates an ExporterFuture implementing a push gateway.
+pub(super) fn new_remote_write(
+    endpoint: Uri,
+    interval: Duration,
+    handle: PrometheusHandle,
+) -> ExporterFuture {
+    Box::pin(async move {
+        let https = hyper_rustls::HttpsConnectorBuilder::new()
+            .with_native_roots()
+            .expect("no native root CA certificates found")
+            .https_or_http()
+            .enable_http1()
+            .build();
+        let client: Client<_, Full<Bytes>> = Client::builder(TokioExecutor::new())
+            .pool_idle_timeout(Duration::from_secs(30))
+            .build(https);
+
+        loop {
+            // Sleep for `interval` amount of time, and then do a push.
+            tokio::time::sleep(interval).await;
+
+            let mut builder = Request::builder();
+
+            let output = handle.render();
+            let result = builder.method(Method::PUT).uri(endpoint.clone()).body(Full::from(output));
+            let req = match result {
+                Ok(req) => req,
+                Err(e) => {
+                    error!("failed to build push gateway request: {}", e);
+                    continue;
+                }
+            };
+
+            match client.request(req).await {
+                Ok(response) => {
+                    if !response.status().is_success() {
+                        let status = response.status();
+                        let status = status.canonical_reason().unwrap_or_else(|| status.as_str());
+                        let body = response
+                            .into_body()
+                            .collect()
+                            .await
+                            .map(Collected::to_bytes)
+                            .map_err(|_| ())
+                            .and_then(|b| String::from_utf8(b[..].to_vec()).map_err(|_| ()))
+                            .unwrap_or_else(|()| String::from("<failed to read response body>"));
+                        error!(
+                            message = "unexpected status after pushing metrics to push gateway",
+                            status,
+                            %body,
+                        );
+                    }
+                }
+                Err(e) => error!("error sending request to push gateway: {:?}", e),
+            }
+        }
+    })
+}
diff --git a/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs b/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
new file mode 100644
index 00000000..09333fa8
--- /dev/null
+++ b/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
@@ -0,0 +1,275 @@
+//! Types and utilities for calling Prometheus remote write API endpoints.
+
+use hyper::{header, Method, Request, Uri};
+
+/// Special label for the name of a metric.
+pub const LABEL_NAME: &str = "__name__";
+pub const CONTENT_TYPE: &str = "application/x-protobuf";
+pub const HEADER_NAME_REMOTE_WRITE_VERSION: &str = "X-Prometheus-Remote-Write-Version";
+pub const REMOTE_WRITE_VERSION_01: &str = "0.1.0";
+
+/// A write request.
+///
+/// .proto:
+/// ```protobuf
+/// message WriteRequest {
+///   repeated TimeSeries timeseries = 1;
+///   // Cortex uses this field to determine the source of the write request.
+///   // We reserve it to avoid any compatibility issues.
+///   reserved  2;
+
+///   // Prometheus uses this field to send metadata, but this is
+///   // omitted from v1 of the spec as it is experimental.
+///   reserved  3;
+/// }
+/// ```
+#[derive(prost::Message, Clone, PartialEq)]
+pub struct WriteRequest {
+    #[prost(message, repeated, tag = "1")]
+    pub timeseries: Vec<TimeSeries>,
+}
+
+impl WriteRequest {
+    /// Prepare the write request for sending.
+    ///
+    /// Ensures that the request conforms to the specification.
+    /// See https://prometheus.io/docs/concepts/remote_write_spec.
+    pub fn sort(&mut self) {
+        for series in &mut self.timeseries {
+            series.sort_labels_and_samples();
+        }
+    }
+
+    pub fn sorted(mut self) -> Self {
+        self.sort();
+        self
+    }
+
+    /// Encode this write request as a protobuf message.
+    ///
+    /// NOTE: The API requires snappy compression, not a raw protobuf message.
+    pub fn encode_proto3(self) -> Vec<u8> {
+        prost::Message::encode_to_vec(&self.sorted())
+    }
+
+    pub fn encode_compressed(self) -> Result<Vec<u8>, snap::Error> {
+        snap::raw::Encoder::new().compress_vec(&self.encode_proto3())
+    }
+
+    /// Parse metrics from the Prometheus text format, and convert them into a
+    /// [`WriteRequest`].
+
+    pub fn from_text_format(
+        text: String,
+    ) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
+        fn samples_to_timeseries(
+            samples: Vec<prometheus_parse::Sample>,
+        ) -> Result<Vec<TimeSeries>, Box<dyn std::error::Error + Send + Sync>> {
+            let mut all_series = std::collections::HashMap::<String, TimeSeries>::new();
+
+            for sample in &samples {
+                let mut labels =
+                    sample.labels.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect::<Vec<_>>();
+
+                labels.push((LABEL_NAME, sample.metric.as_str()));
+
+                labels.sort_by(|a, b| a.0.cmp(b.0));
+
+                let mut ident = sample.metric.clone();
+                ident.push_str("_$$_");
+                for (k, v) in &labels {
+                    ident.push_str(k);
+                    ident.push('=');
+                    ident.push_str(v);
+                }
+
+                let series = all_series.entry(ident).or_insert_with(|| {
+                    let labels = labels
+                        .iter()
+                        .map(|(k, v)| Label { name: k.to_string(), value: v.to_string() })
+                        .collect::<Vec<_>>();
+
+                    TimeSeries { labels, samples: vec![] }
+                });
+
+                let value = match sample.value {
+                    prometheus_parse::Value::Counter(v) => v,
+                    prometheus_parse::Value::Gauge(v) => v,
+                    prometheus_parse::Value::Histogram(_) => {
+                        Err("histogram not supported yet".to_string())?
+                    }
+                    prometheus_parse::Value::Summary(_) => {
+                        Err("summary not supported yet".to_string())?
+                    }
+                    prometheus_parse::Value::Untyped(v) => v,
+                };
+
+                series
+                    .samples
+                    .push(Sample { value, timestamp: sample.timestamp.timestamp_millis() });
+            }
+
+            Ok(all_series.into_values().collect())
+        }
+
+        let iter = text.trim().lines().map(|x| Ok(x.to_string()));
+        let parsed = prometheus_parse::Scrape::parse(iter)
+            .map_err(|err| format!("could not parse input as Prometheus text format: {err}"))?;
+
+        let mut series = samples_to_timeseries(parsed.samples)?;
+        series.sort_by(|a, b| {
+            let name_a = a.labels.iter().find(|x| x.name == LABEL_NAME).unwrap();
+            let name_b = b.labels.iter().find(|x| x.name == LABEL_NAME).unwrap();
+            name_a.value.cmp(&name_b.value)
+        });
+
+        let s = Self { timeseries: series };
+
+        Ok(s.sorted())
+    }
+
+    /// Build a fully prepared HTTP request that an be sent to a remote write endpoint.
+    pub fn build_http_request(
+        self,
+        endpoint: &Uri,
+        user_agent: &str,
+    ) -> Result<Request<Vec<u8>>, Box<dyn std::error::Error + Send + Sync>> {
+        let req = Request::builder()
+            .method(Method::POST)
+            .uri(endpoint)
+            .header(header::CONTENT_TYPE, CONTENT_TYPE)
+            .header(HEADER_NAME_REMOTE_WRITE_VERSION, REMOTE_WRITE_VERSION_01)
+            .header(header::CONTENT_ENCODING, "snappy")
+            .header(header::USER_AGENT, user_agent)
+            .body(self.encode_compressed()?)?;
+
+        Ok(req)
+    }
+}
+
+/// A time series.
+///
+/// .proto:
+/// ```protobuf
+/// message TimeSeries {
+///   repeated Label labels   = 1;
+///   repeated Sample samples = 2;
+/// }
+/// ```
+#[derive(prost::Message, Clone, PartialEq)]
+pub struct TimeSeries {
+    #[prost(message, repeated, tag = "1")]
+    pub labels: Vec<Label>,
+    #[prost(message, repeated, tag = "2")]
+    pub samples: Vec<Sample>,
+}
+
+impl TimeSeries {
+    /// Sort labels by name, and the samples by timestamp.
+    ///
+    /// Required by the specification.
+    pub fn sort_labels_and_samples(&mut self) {
+        self.labels.sort_by(|a, b| a.name.cmp(&b.name));
+        self.samples.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
+    }
+}
+
+/// A label.
+///
+/// .proto:
+/// ```protobuf
+/// message Label {
+///   string name  = 1;
+///   string value = 2;
+/// }
+/// ```
+#[derive(prost::Message, Clone, Hash, PartialEq, Eq)]
+pub struct Label {
+    #[prost(string, tag = "1")]
+    pub name: String,
+    #[prost(string, tag = "2")]
+    pub value: String,
+}
+
+/// A sample.
+///
+/// .proto:
+/// ```protobuf
+/// message Sample {
+///   double value    = 1;
+///   int64 timestamp = 2;
+/// }
+/// ```
+#[derive(prost::Message, Clone, PartialEq)]
+pub struct Sample {
+    #[prost(double, tag = "1")]
+    pub value: f64,
+    #[prost(int64, tag = "2")]
+    pub timestamp: i64,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn test_name() {
+        let input = r#"
+# TYPE mycounter counter
+# TYPE mygauge gauge
+
+mygauge 100 100
+http_requests_total{method="post",code="200"} 1027 1395066363000
+mycounter 100 100
+alpha 10 1000
+http_requests_total{method="post",code="200"} 50 1000
+    "#;
+
+        let req = WriteRequest::from_text_format(input.to_string()).unwrap();
+
+        assert_eq!(
+            req,
+            WriteRequest {
+                timeseries: vec![
+                    TimeSeries {
+                        labels: vec![Label {
+                            name: LABEL_NAME.to_string(),
+                            value: "alpha".to_string()
+                        },],
+                        samples: vec![Sample { value: 10.0, timestamp: 1000 },]
+                    },
+                    TimeSeries {
+                        labels: vec![
+                            Label {
+                                name: LABEL_NAME.to_string(),
+                                value: "http_requests_total".to_string()
+                            },
+                            Label { name: "code".to_string(), value: "200".to_string() },
+                            Label { name: "method".to_string(), value: "post".to_string() },
+                        ],
+                        samples: vec![
+                            Sample { value: 50.0, timestamp: 1000 },
+                            Sample { value: 1027.0, timestamp: 1395066363000 },
+                        ]
+                    },
+                    TimeSeries {
+                        labels: vec![Label {
+                            name: LABEL_NAME.to_string(),
+                            value: "mycounter".to_string()
+                        },],
+                        samples: vec![Sample { value: 100.0, timestamp: 100 }],
+                    },
+                    TimeSeries {
+                        labels: vec![Label {
+                            name: LABEL_NAME.to_string(),
+                            value: "mygauge".to_string()
+                        },],
+                        samples: vec![Sample { value: 100.0, timestamp: 100 }],
+                    },
+                ]
+            }
+        );
+
+        let _x = req.clone().encode_proto3();
+        let _y = req.encode_compressed();
+    }
+}

From f93d06cac8f84d2b9b3e23cb8dc697dc88fe36dc Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Sun, 3 Nov 2024 15:41:09 +0800
Subject: [PATCH 02/11] 1

---
 .../examples/prometheus_remote_write.rs       | 73 +++++++++++++++++++
 metrics-exporter-prometheus/src/common.rs     |  4 +-
 .../src/exporter/builder.rs                   | 35 +++++++++
 .../src/exporter/mod.rs                       |  7 ++
 .../src/exporter/remote_write.rs              | 26 ++++---
 .../src/exporter/remote_write_proto.rs        | 15 ++--
 6 files changed, 141 insertions(+), 19 deletions(-)
 create mode 100644 metrics-exporter-prometheus/examples/prometheus_remote_write.rs

diff --git a/metrics-exporter-prometheus/examples/prometheus_remote_write.rs b/metrics-exporter-prometheus/examples/prometheus_remote_write.rs
new file mode 100644
index 00000000..983eb590
--- /dev/null
+++ b/metrics-exporter-prometheus/examples/prometheus_remote_write.rs
@@ -0,0 +1,73 @@
+/// Make sure to run this example with `--features remote-write` to properly enable remote write support.
+#[allow(unused_imports)]
+use std::thread;
+use std::time::Duration;
+
+#[allow(unused_imports)]
+use metrics::{counter, gauge, histogram};
+use metrics::{describe_counter, describe_histogram};
+#[allow(unused_imports)]
+use metrics_exporter_prometheus::PrometheusBuilder;
+#[allow(unused_imports)]
+use metrics_util::MetricKindMask;
+
+use quanta::Clock;
+use rand::{thread_rng, Rng};
+
+fn main() {
+    tracing_subscriber::fmt::init();
+
+    PrometheusBuilder::new()
+        .with_remote_write(
+            "http://127.0.0.1:9091/metrics/job/example",
+            Duration::from_secs(10),
+            "test-agent",
+        )
+        .expect("remote write endpoint should be valid")
+        .idle_timeout(
+            MetricKindMask::COUNTER | MetricKindMask::HISTOGRAM,
+            Some(Duration::from_secs(10)),
+        )
+        .install()
+        .expect("failed to install Prometheus recorder");
+
+    // We register these metrics, which gives us a chance to specify a description for them.  The
+    // Prometheus exporter records this description and adds it as HELP text when the endpoint is
+    // scraped.
+    //
+    // Registering metrics ahead of using them is not required, but is the only way to specify the
+    // description of a metric.
+    describe_counter!("tcp_server_loops", "The iterations of the TCP server event loop so far.");
+    describe_histogram!(
+        "tcp_server_loop_delta_secs",
+        "The time taken for iterations of the TCP server event loop."
+    );
+
+    let clock = Clock::new();
+    let mut last = None;
+
+    counter!("idle_metric").increment(1);
+    gauge!("testing").set(42.0);
+
+    // Loop over and over, pretending to do some work.
+    loop {
+        counter!("tcp_server_loops", "system" => "foo").increment(1);
+
+        if let Some(t) = last {
+            let delta: Duration = clock.now() - t;
+            histogram!("tcp_server_loop_delta_secs", "system" => "foo").record(delta);
+        }
+
+        let increment_gauge = thread_rng().gen_bool(0.75);
+        let gauge = gauge!("lucky_iterations");
+        if increment_gauge {
+            gauge.increment(1.0);
+        } else {
+            gauge.decrement(1.0);
+        }
+
+        last = Some(clock.now());
+
+        thread::sleep(Duration::from_millis(750));
+    }
+}
diff --git a/metrics-exporter-prometheus/src/common.rs b/metrics-exporter-prometheus/src/common.rs
index 51f9bad9..8f422a2f 100644
--- a/metrics-exporter-prometheus/src/common.rs
+++ b/metrics-exporter-prometheus/src/common.rs
@@ -63,7 +63,9 @@ pub enum BuildError {
     /// The given push gateway endpoint is not a valid URI.
     #[error("push gateway endpoint is not valid: {0}")]
     InvalidPushGatewayEndpoint(String),
-
+    /// The given push gateway endpoint is not a valid URI.
+    #[error("remote write endpoint is not valid: {0}")]
+    InvalidRemoteWriteEndpoint(String),
     /// No exporter configuration was present.
     ///
     /// This generally only occurs when HTTP listener support is disabled, but no push gateway
diff --git a/metrics-exporter-prometheus/src/exporter/builder.rs b/metrics-exporter-prometheus/src/exporter/builder.rs
index fc019b04..9c5416e3 100644
--- a/metrics-exporter-prometheus/src/exporter/builder.rs
+++ b/metrics-exporter-prometheus/src/exporter/builder.rs
@@ -138,6 +138,37 @@ impl PrometheusBuilder {
         Ok(self)
     }
 
+    /// Configures the exporter to push periodic requests to endpoint by [remote write protocol](https://prometheus.io/docs/specs/remote_write_spec/).
+    ///
+    /// Running in remote write mode is mutually exclusive with the HTTP listener/push gateway i.e. enabling the remote write will
+    /// disable the HTTP listener/push gateway, and vise versa.
+    ///
+    /// Defaults to disabled.
+    ///
+    /// ## Errors
+    ///
+    /// If the given endpoint cannot be parsed into a valid URI, an error variant will be returned describing the error.
+    ///
+    #[cfg(feature = "push-gateway")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "push-gateway")))]
+    pub fn with_remote_write<T>(
+        mut self,
+        endpoint: T,
+        interval: Duration,
+        user_agent: &str,
+    ) -> Result<Self, BuildError>
+    where
+        T: AsRef<str>,
+    {
+        self.exporter_config = ExporterConfig::RemoteWrite {
+            endpoint: Uri::try_from(endpoint.as_ref())
+                .map_err(|e| BuildError::InvalidPushGatewayEndpoint(e.to_string()))?,
+            interval,
+            user_agent: user_agent.to_string(),
+        };
+        Ok(self)
+    }
+
     /// Configures the exporter to expose an HTTP listener that functions as a [scrape endpoint], listening on a Unix
     /// Domain socket at the given path
     ///
@@ -486,6 +517,10 @@ impl PrometheusBuilder {
                         endpoint, interval, username, password, handle,
                     )
                 }
+                #[cfg(feature = "remote-write")]
+                ExporterConfig::RemoteWrite { endpoint, interval, user_agent } => {
+                    super::remote_write::new_remote_write(endpoint, interval, handle, &user_agent)
+                }
             },
         ))
     }
diff --git a/metrics-exporter-prometheus/src/exporter/mod.rs b/metrics-exporter-prometheus/src/exporter/mod.rs
index 490daa64..6afd8a66 100644
--- a/metrics-exporter-prometheus/src/exporter/mod.rs
+++ b/metrics-exporter-prometheus/src/exporter/mod.rs
@@ -48,6 +48,11 @@ enum ExporterConfig {
         password: Option<String>,
     },
 
+    // Run a remote write task sending to the given `endpoint` after `interval` time has elapsed,
+    // infinitely.
+    #[cfg(feature = "remote-write")]
+    RemoteWrite { endpoint: Uri, interval: Duration, user_agent: String },
+
     #[allow(dead_code)]
     Unconfigured,
 }
@@ -60,6 +65,8 @@ impl ExporterConfig {
             Self::HttpListener { .. } => "http-listener",
             #[cfg(feature = "push-gateway")]
             Self::PushGateway { .. } => "push-gateway",
+            #[cfg(feature = "remote-write")]
+            Self::RemoteWrite { .. } => "remote-write",
             Self::Unconfigured => "unconfigured,",
         }
     }
diff --git a/metrics-exporter-prometheus/src/exporter/remote_write.rs b/metrics-exporter-prometheus/src/exporter/remote_write.rs
index a0ce0544..5e9e38ab 100644
--- a/metrics-exporter-prometheus/src/exporter/remote_write.rs
+++ b/metrics-exporter-prometheus/src/exporter/remote_write.rs
@@ -1,20 +1,22 @@
 use std::time::Duration;
 
 use http_body_util::{BodyExt, Collected, Full};
-use hyper::{body::Bytes, Method, Request, Uri};
+use hyper::{body::Bytes, Uri};
 use hyper_util::{client::legacy::Client, rt::TokioExecutor};
 use tracing::error;
 
 use crate::PrometheusHandle;
 
-use super::ExporterFuture;
+use super::{remote_write_proto::WriteRequest, ExporterFuture};
 
 // Creates an ExporterFuture implementing a push gateway.
 pub(super) fn new_remote_write(
     endpoint: Uri,
     interval: Duration,
     handle: PrometheusHandle,
+    user_agent: &str,
 ) -> ExporterFuture {
+    let user_agent = user_agent.to_string();
     Box::pin(async move {
         let https = hyper_rustls::HttpsConnectorBuilder::new()
             .with_native_roots()
@@ -30,18 +32,22 @@ pub(super) fn new_remote_write(
             // Sleep for `interval` amount of time, and then do a push.
             tokio::time::sleep(interval).await;
 
-            let mut builder = Request::builder();
-
             let output = handle.render();
-            let result = builder.method(Method::PUT).uri(endpoint.clone()).body(Full::from(output));
-            let req = match result {
+            let binary = match WriteRequest::from_text_format(output) {
                 Ok(req) => req,
-                Err(e) => {
-                    error!("failed to build push gateway request: {}", e);
+                Err(err) => {
+                    error!("failed to build output to remote write request: {}", err);
                     continue;
                 }
             };
 
+            let req = match binary.build_http_request(&endpoint, &user_agent) {
+                Ok(req) => req,
+                Err(err) => {
+                    error!("failed to build http remote write request {}", err);
+                    continue;
+                }
+            };
             match client.request(req).await {
                 Ok(response) => {
                     if !response.status().is_success() {
@@ -56,13 +62,13 @@ pub(super) fn new_remote_write(
                             .and_then(|b| String::from_utf8(b[..].to_vec()).map_err(|_| ()))
                             .unwrap_or_else(|()| String::from("<failed to read response body>"));
                         error!(
-                            message = "unexpected status after pushing metrics to push gateway",
+                            message = "unexpected status after pushing metrics to remote write",
                             status,
                             %body,
                         );
                     }
                 }
-                Err(e) => error!("error sending request to push gateway: {:?}", e),
+                Err(e) => error!("error sending request to remote write {}: {:?}", endpoint, e),
             }
         }
     })
diff --git a/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs b/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
index 09333fa8..95f2c607 100644
--- a/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
+++ b/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
@@ -1,6 +1,7 @@
 //! Types and utilities for calling Prometheus remote write API endpoints.
-
-use hyper::{header, Method, Request, Uri};
+// Copy from https://github.com/theduke/prom-write
+use http_body_util::Full;
+use hyper::{body::Bytes, header, Method, Request, Uri};
 
 /// Special label for the name of a metric.
 pub const LABEL_NAME: &str = "__name__";
@@ -46,12 +47,11 @@ impl WriteRequest {
     }
 
     /// Encode this write request as a protobuf message.
-    ///
-    /// NOTE: The API requires snappy compression, not a raw protobuf message.
     pub fn encode_proto3(self) -> Vec<u8> {
         prost::Message::encode_to_vec(&self.sorted())
     }
-
+    /// Encode this write request as a compressed protobuf message.
+    /// NOTE: The API requires snappy compression, not a raw protobuf message.
     pub fn encode_compressed(self) -> Result<Vec<u8>, snap::Error> {
         snap::raw::Encoder::new().compress_vec(&self.encode_proto3())
     }
@@ -133,7 +133,7 @@ impl WriteRequest {
         self,
         endpoint: &Uri,
         user_agent: &str,
-    ) -> Result<Request<Vec<u8>>, Box<dyn std::error::Error + Send + Sync>> {
+    ) -> Result<Request<Full<Bytes>>, Box<dyn std::error::Error + Send + Sync>> {
         let req = Request::builder()
             .method(Method::POST)
             .uri(endpoint)
@@ -141,8 +141,7 @@ impl WriteRequest {
             .header(HEADER_NAME_REMOTE_WRITE_VERSION, REMOTE_WRITE_VERSION_01)
             .header(header::CONTENT_ENCODING, "snappy")
             .header(header::USER_AGENT, user_agent)
-            .body(self.encode_compressed()?)?;
-
+            .body(Full::new(self.encode_compressed()?.into()))?;
         Ok(req)
     }
 }

From 7bd7ad1d3bad6b0c860b88e7b148db1136a9cf35 Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Sun, 3 Nov 2024 15:45:18 +0800
Subject: [PATCH 03/11] 1

---
 metrics-exporter-prometheus/src/exporter/push_gateway.rs | 1 +
 1 file changed, 1 insertion(+)

diff --git a/metrics-exporter-prometheus/src/exporter/push_gateway.rs b/metrics-exporter-prometheus/src/exporter/push_gateway.rs
index a1c2a4e7..e5ef04ed 100644
--- a/metrics-exporter-prometheus/src/exporter/push_gateway.rs
+++ b/metrics-exporter-prometheus/src/exporter/push_gateway.rs
@@ -40,6 +40,7 @@ pub(super) fn new_push_gateway(
             }
 
             let output = handle.render();
+            println!("output {}", output);
             let result = builder.method(Method::PUT).uri(endpoint.clone()).body(Full::from(output));
             let req = match result {
                 Ok(req) => req,

From f7bc760f98c6bf9bbc7632b8cdd7ed00baa1408d Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Sun, 3 Nov 2024 15:52:10 +0800
Subject: [PATCH 04/11] 1

---
 metrics-exporter-prometheus/Cargo.toml                   | 2 +-
 metrics-exporter-prometheus/src/exporter/builder.rs      | 6 +++---
 metrics-exporter-prometheus/src/exporter/push_gateway.rs | 1 -
 metrics-exporter-prometheus/src/exporter/remote_write.rs | 2 +-
 4 files changed, 5 insertions(+), 6 deletions(-)

diff --git a/metrics-exporter-prometheus/Cargo.toml b/metrics-exporter-prometheus/Cargo.toml
index 21c23385..f69c83bd 100644
--- a/metrics-exporter-prometheus/Cargo.toml
+++ b/metrics-exporter-prometheus/Cargo.toml
@@ -22,7 +22,7 @@ async-runtime = ["tokio", "hyper-util/tokio"]
 http-listener = ["async-runtime", "ipnet", "tracing", "_hyper-server"]
 uds-listener = ["http-listener"]
 push-gateway = ["async-runtime", "tracing", "_hyper-client"]
-remote-write = ["_hyper-client","async-runtime","dep:prost","dep:snap","dep:prometheus-parse","dep:prost"]
+remote-write = ["_hyper-client","async-runtime","dep:prost","dep:snap","dep:prometheus-parse"]
 _hyper-server = ["http-body-util", "hyper/server", "hyper-util/server-auto"]
 _hyper-client = [
     "http-body-util",
diff --git a/metrics-exporter-prometheus/src/exporter/builder.rs b/metrics-exporter-prometheus/src/exporter/builder.rs
index 9c5416e3..ac2930a3 100644
--- a/metrics-exporter-prometheus/src/exporter/builder.rs
+++ b/metrics-exporter-prometheus/src/exporter/builder.rs
@@ -149,8 +149,8 @@ impl PrometheusBuilder {
     ///
     /// If the given endpoint cannot be parsed into a valid URI, an error variant will be returned describing the error.
     ///
-    #[cfg(feature = "push-gateway")]
-    #[cfg_attr(docsrs, doc(cfg(feature = "push-gateway")))]
+    #[cfg(feature = "remote-write")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "remote-write")))]
     pub fn with_remote_write<T>(
         mut self,
         endpoint: T,
@@ -162,7 +162,7 @@ impl PrometheusBuilder {
     {
         self.exporter_config = ExporterConfig::RemoteWrite {
             endpoint: Uri::try_from(endpoint.as_ref())
-                .map_err(|e| BuildError::InvalidPushGatewayEndpoint(e.to_string()))?,
+                .map_err(|e| BuildError::InvalidRemoteWriteEndpoint(e.to_string()))?,
             interval,
             user_agent: user_agent.to_string(),
         };
diff --git a/metrics-exporter-prometheus/src/exporter/push_gateway.rs b/metrics-exporter-prometheus/src/exporter/push_gateway.rs
index e5ef04ed..a1c2a4e7 100644
--- a/metrics-exporter-prometheus/src/exporter/push_gateway.rs
+++ b/metrics-exporter-prometheus/src/exporter/push_gateway.rs
@@ -40,7 +40,6 @@ pub(super) fn new_push_gateway(
             }
 
             let output = handle.render();
-            println!("output {}", output);
             let result = builder.method(Method::PUT).uri(endpoint.clone()).body(Full::from(output));
             let req = match result {
                 Ok(req) => req,
diff --git a/metrics-exporter-prometheus/src/exporter/remote_write.rs b/metrics-exporter-prometheus/src/exporter/remote_write.rs
index 5e9e38ab..1b024468 100644
--- a/metrics-exporter-prometheus/src/exporter/remote_write.rs
+++ b/metrics-exporter-prometheus/src/exporter/remote_write.rs
@@ -9,7 +9,7 @@ use crate::PrometheusHandle;
 
 use super::{remote_write_proto::WriteRequest, ExporterFuture};
 
-// Creates an ExporterFuture implementing a push gateway.
+// Creates an ExporterFuture implementing a remote write.
 pub(super) fn new_remote_write(
     endpoint: Uri,
     interval: Duration,

From 747c517089d37c390b42eea7b8b915cc57e6c90c Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Sun, 3 Nov 2024 15:56:45 +0800
Subject: [PATCH 05/11] 1

---
 .../src/exporter/remote_write_proto.rs                        | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs b/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
index 95f2c607..8c80daa6 100644
--- a/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
+++ b/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
@@ -35,13 +35,13 @@ impl WriteRequest {
     ///
     /// Ensures that the request conforms to the specification.
     /// See https://prometheus.io/docs/concepts/remote_write_spec.
-    pub fn sort(&mut self) {
+    fn sort(&mut self) {
         for series in &mut self.timeseries {
             series.sort_labels_and_samples();
         }
     }
 
-    pub fn sorted(mut self) -> Self {
+    fn sorted(mut self) -> Self {
         self.sort();
         self
     }

From 2ef8c565396d99f442178665614db4df075de2ae Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Sun, 3 Nov 2024 15:59:35 +0800
Subject: [PATCH 06/11] 1

---
 metrics-exporter-prometheus/Cargo.toml | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/metrics-exporter-prometheus/Cargo.toml b/metrics-exporter-prometheus/Cargo.toml
index f69c83bd..475a82dd 100644
--- a/metrics-exporter-prometheus/Cargo.toml
+++ b/metrics-exporter-prometheus/Cargo.toml
@@ -67,6 +67,10 @@ tracing-subscriber = { workspace = true, features = ["fmt"] }
 name = "prometheus_push_gateway"
 required-features = ["push-gateway"]
 
+[[example]]
+name = "prometheus_remote_write"
+required-features = ["remote-write"]
+
 [[example]]
 name = "prometheus_server"
 required-features = ["http-listener"]

From 5d0e81ea78696a871a2e050988f23d427746eb36 Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Mon, 4 Nov 2024 13:16:25 +0800
Subject: [PATCH 07/11] 1

---
 metrics-exporter-prometheus/src/exporter/mod.rs  |  2 --
 .../src/exporter/remote_write.rs                 | 11 ++---------
 metrics-exporter-prometheus/src/lib.rs           |  7 ++++---
 metrics-exporter-prometheus/src/recorder.rs      | 16 ++++++++++++++--
 .../src/{exporter => }/remote_write_proto.rs     | 14 +++++++++-----
 5 files changed, 29 insertions(+), 21 deletions(-)
 rename metrics-exporter-prometheus/src/{exporter => }/remote_write_proto.rs (95%)

diff --git a/metrics-exporter-prometheus/src/exporter/mod.rs b/metrics-exporter-prometheus/src/exporter/mod.rs
index 6afd8a66..09d3f2e9 100644
--- a/metrics-exporter-prometheus/src/exporter/mod.rs
+++ b/metrics-exporter-prometheus/src/exporter/mod.rs
@@ -80,7 +80,5 @@ mod push_gateway;
 
 #[cfg(feature = "remote-write")]
 mod remote_write;
-#[cfg(feature = "remote-write")]
-mod remote_write_proto;
 
 pub(crate) mod builder;
diff --git a/metrics-exporter-prometheus/src/exporter/remote_write.rs b/metrics-exporter-prometheus/src/exporter/remote_write.rs
index 1b024468..9f44e4b9 100644
--- a/metrics-exporter-prometheus/src/exporter/remote_write.rs
+++ b/metrics-exporter-prometheus/src/exporter/remote_write.rs
@@ -7,7 +7,7 @@ use tracing::error;
 
 use crate::PrometheusHandle;
 
-use super::{remote_write_proto::WriteRequest, ExporterFuture};
+use super::ExporterFuture;
 
 // Creates an ExporterFuture implementing a remote write.
 pub(super) fn new_remote_write(
@@ -32,14 +32,7 @@ pub(super) fn new_remote_write(
             // Sleep for `interval` amount of time, and then do a push.
             tokio::time::sleep(interval).await;
 
-            let output = handle.render();
-            let binary = match WriteRequest::from_text_format(output) {
-                Ok(req) => req,
-                Err(err) => {
-                    error!("failed to build output to remote write request: {}", err);
-                    continue;
-                }
-            };
+            let binary = handle.render_remote_write_format();
 
             let req = match binary.build_http_request(&endpoint, &user_agent) {
                 Ok(req) => req,
diff --git a/metrics-exporter-prometheus/src/lib.rs b/metrics-exporter-prometheus/src/lib.rs
index 2240808c..213b4436 100644
--- a/metrics-exporter-prometheus/src/lib.rs
+++ b/metrics-exporter-prometheus/src/lib.rs
@@ -120,10 +120,11 @@ pub use distribution::{Distribution, DistributionBuilder};
 
 mod exporter;
 pub use self::exporter::builder::PrometheusBuilder;
-#[cfg(any(feature = "http-listener", feature = "push-gateway"))]
-#[cfg_attr(docsrs, doc(cfg(any(feature = "http-listener", feature = "push-gateway"))))]
+#[cfg(any(feature = "http-listener", feature = "push-gateway", feature = "remote-write"))]
+#[cfg_attr(docsrs, doc(cfg(any(feature = "http-listener", feature = "push-gateway", feature = "remote-write"))))]
 pub use self::exporter::ExporterFuture;
-
+#[cfg(feature = "remote-write")]
+mod remote_write_proto;
 pub mod formatting;
 mod recorder;
 
diff --git a/metrics-exporter-prometheus/src/recorder.rs b/metrics-exporter-prometheus/src/recorder.rs
index 7d49f688..35549888 100644
--- a/metrics-exporter-prometheus/src/recorder.rs
+++ b/metrics-exporter-prometheus/src/recorder.rs
@@ -26,7 +26,7 @@ pub(crate) struct Inner {
 }
 
 impl Inner {
-    fn get_recent_metrics(&self) -> Snapshot {
+    pub(crate) fn get_recent_metrics(&self) -> Snapshot {
         let mut counters = HashMap::new();
         let counter_handles = self.registry.get_counter_handles();
         for (key, counter) in counter_handles {
@@ -108,7 +108,12 @@ impl Inner {
             histogram.get_inner().clear_with(|samples| entry.record_samples(samples));
         }
     }
-
+    /// Render metric to [Remote-Write format](https://prometheus.io/docs/specs/remote_write_spec/)
+    #[cfg(feature = "remote-write")]
+    fn render_remote_write_format(&self) -> crate::remote_write_proto::WriteRequest {
+        use crate::remote_write_proto::WriteRequest;
+        WriteRequest::from_raw(self)
+    }
     fn render(&self) -> String {
         let Snapshot { mut counters, mut distributions, mut gauges } = self.get_recent_metrics();
 
@@ -289,6 +294,13 @@ impl PrometheusHandle {
         self.inner.render()
     }
 
+    /// Takes a snapshot of the metrics held by the recorder and generates a payload conforming to
+    /// the Prometheus remote write format.
+    #[cfg(feature = "remote-write")]
+    pub fn render_remote_write_format(&self) -> crate::remote_write_proto::WriteRequest {
+        self.inner.render_remote_write_format()
+    }
+
     /// Performs upkeeping operations to ensure metrics held by recorder are up-to-date and do not
     /// grow unboundedly.
     pub fn run_upkeep(&self) {
diff --git a/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs b/metrics-exporter-prometheus/src/remote_write_proto.rs
similarity index 95%
rename from metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
rename to metrics-exporter-prometheus/src/remote_write_proto.rs
index 8c80daa6..b4128bb7 100644
--- a/metrics-exporter-prometheus/src/exporter/remote_write_proto.rs
+++ b/metrics-exporter-prometheus/src/remote_write_proto.rs
@@ -3,6 +3,8 @@
 use http_body_util::Full;
 use hyper::{body::Bytes, header, Method, Request, Uri};
 
+use crate::{common::Snapshot, recorder::Inner};
+
 /// Special label for the name of a metric.
 pub const LABEL_NAME: &str = "__name__";
 pub const CONTENT_TYPE: &str = "application/x-protobuf";
@@ -56,12 +58,14 @@ impl WriteRequest {
         snap::raw::Encoder::new().compress_vec(&self.encode_proto3())
     }
 
+    /// Parse metrics from inner metric object, and convert them into a [`WriteRequest`]
+    pub fn from_raw(inner: &Inner) -> Self {
+        let Snapshot { mut counters, mut distributions, mut gauges } = inner.get_recent_metrics();
+        todo!()
+    }
     /// Parse metrics from the Prometheus text format, and convert them into a
     /// [`WriteRequest`].
-
-    pub fn from_text_format(
-        text: String,
-    ) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
+    fn from_text_format(text: String) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
         fn samples_to_timeseries(
             samples: Vec<prometheus_parse::Sample>,
         ) -> Result<Vec<TimeSeries>, Box<dyn std::error::Error + Send + Sync>> {
@@ -123,7 +127,7 @@ impl WriteRequest {
             name_a.value.cmp(&name_b.value)
         });
 
-        let s = Self { timeseries: series };
+        let s = WriteRequest { timeseries: series };
 
         Ok(s.sorted())
     }

From 965b9385e464e2008f4f0d39443465d6886e200e Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Mon, 4 Nov 2024 13:22:56 +0800
Subject: [PATCH 08/11] 1

---
 .../src/remote_write_proto.rs                 | 93 ++++++++++++++++++-
 1 file changed, 91 insertions(+), 2 deletions(-)

diff --git a/metrics-exporter-prometheus/src/remote_write_proto.rs b/metrics-exporter-prometheus/src/remote_write_proto.rs
index b4128bb7..0ebc679b 100644
--- a/metrics-exporter-prometheus/src/remote_write_proto.rs
+++ b/metrics-exporter-prometheus/src/remote_write_proto.rs
@@ -1,4 +1,6 @@
 //! Types and utilities for calling Prometheus remote write API endpoints.
+use std::sync::PoisonError;
+
 // Copy from https://github.com/theduke/prom-write
 use http_body_util::Full;
 use hyper::{body::Bytes, header, Method, Request, Uri};
@@ -59,9 +61,96 @@ impl WriteRequest {
     }
 
     /// Parse metrics from inner metric object, and convert them into a [`WriteRequest`]
-    pub fn from_raw(inner: &Inner) -> Self {
+    pub(super) fn from_raw(inner: &Inner) -> Self {
         let Snapshot { mut counters, mut distributions, mut gauges } = inner.get_recent_metrics();
-        todo!()
+        let descriptions = inner.descriptions.read().unwrap_or_else(PoisonError::into_inner);
+        let mut req = WriteRequest::default();
+        let mut all_series = std::collections::HashMap::<String, TimeSeries>::new();
+        for (name, mut by_labels) in counters.drain() {
+            if let Some(desc) = descriptions.get(name.as_str()) {
+
+            }
+
+            write_type_line(&mut output, name.as_str(), "counter");
+            for (labels, value) in by_labels.drain() {
+                write_metric_line::<&str, u64>(&mut output, &name, None, &labels, None, value);
+            }
+            output.push('\n');
+        }
+        for (name, mut by_labels) in gauges.drain() {
+            if let Some(desc) = descriptions.get(name.as_str()) {
+                write_help_line(&mut output, name.as_str(), desc);
+            }
+
+            write_type_line(&mut output, name.as_str(), "gauge");
+            for (labels, value) in by_labels.drain() {
+                write_metric_line::<&str, f64>(&mut output, &name, None, &labels, None, value);
+            }
+            output.push('\n');
+        }
+        for (name, mut by_labels) in distributions.drain() {
+            if let Some(desc) = descriptions.get(name.as_str()) {
+                write_help_line(&mut output, name.as_str(), desc);
+            }
+
+            let distribution_type = self.distribution_builder.get_distribution_type(name.as_str());
+            write_type_line(&mut output, name.as_str(), distribution_type);
+            for (labels, distribution) in by_labels.drain(..) {
+                let (sum, count) = match distribution {
+                    Distribution::Summary(summary, quantiles, sum) => {
+                        let snapshot = summary.snapshot(Instant::now());
+                        for quantile in quantiles.iter() {
+                            let value = snapshot.quantile(quantile.value()).unwrap_or(0.0);
+                            write_metric_line(
+                                &mut output,
+                                &name,
+                                None,
+                                &labels,
+                                Some(("quantile", quantile.value())),
+                                value,
+                            );
+                        }
+
+                        (sum, summary.count() as u64)
+                    }
+                    Distribution::Histogram(histogram) => {
+                        for (le, count) in histogram.buckets() {
+                            write_metric_line(
+                                &mut output,
+                                &name,
+                                Some("bucket"),
+                                &labels,
+                                Some(("le", le)),
+                                count,
+                            );
+                        }
+                        write_metric_line(
+                            &mut output,
+                            &name,
+                            Some("bucket"),
+                            &labels,
+                            Some(("le", "+Inf")),
+                            histogram.count(),
+                        );
+
+                        (histogram.sum(), histogram.count())
+                    }
+                };
+
+                write_metric_line::<&str, f64>(&mut output, &name, Some("sum"), &labels, None, sum);
+                write_metric_line::<&str, u64>(
+                    &mut output,
+                    &name,
+                    Some("count"),
+                    &labels,
+                    None,
+                    count,
+                );
+            }
+
+            output.push('\n');
+        }
+        req
     }
     /// Parse metrics from the Prometheus text format, and convert them into a
     /// [`WriteRequest`].

From a9aa142cc7f264e46c33803707bf4558a3ad6d77 Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Mon, 4 Nov 2024 13:31:02 +0800
Subject: [PATCH 09/11] 1

---
 .../src/remote_write_proto.rs                 | 70 +++----------------
 1 file changed, 10 insertions(+), 60 deletions(-)

diff --git a/metrics-exporter-prometheus/src/remote_write_proto.rs b/metrics-exporter-prometheus/src/remote_write_proto.rs
index 0ebc679b..1c4020e7 100644
--- a/metrics-exporter-prometheus/src/remote_write_proto.rs
+++ b/metrics-exporter-prometheus/src/remote_write_proto.rs
@@ -4,8 +4,9 @@ use std::sync::PoisonError;
 // Copy from https://github.com/theduke/prom-write
 use http_body_util::Full;
 use hyper::{body::Bytes, header, Method, Request, Uri};
+use quanta::Instant;
 
-use crate::{common::Snapshot, recorder::Inner};
+use crate::{common::Snapshot, recorder::Inner, Distribution};
 
 /// Special label for the name of a metric.
 pub const LABEL_NAME: &str = "__name__";
@@ -67,88 +68,37 @@ impl WriteRequest {
         let mut req = WriteRequest::default();
         let mut all_series = std::collections::HashMap::<String, TimeSeries>::new();
         for (name, mut by_labels) in counters.drain() {
-            if let Some(desc) = descriptions.get(name.as_str()) {
+            if let Some(desc) = descriptions.get(name.as_str()) {}
 
-            }
-
-            write_type_line(&mut output, name.as_str(), "counter");
-            for (labels, value) in by_labels.drain() {
-                write_metric_line::<&str, u64>(&mut output, &name, None, &labels, None, value);
-            }
-            output.push('\n');
+            for (labels, value) in by_labels.drain() {}
         }
         for (name, mut by_labels) in gauges.drain() {
-            if let Some(desc) = descriptions.get(name.as_str()) {
-                write_help_line(&mut output, name.as_str(), desc);
-            }
+            if let Some(desc) = descriptions.get(name.as_str()) {}
 
-            write_type_line(&mut output, name.as_str(), "gauge");
-            for (labels, value) in by_labels.drain() {
-                write_metric_line::<&str, f64>(&mut output, &name, None, &labels, None, value);
-            }
-            output.push('\n');
+            for (labels, value) in by_labels.drain() {}
         }
         for (name, mut by_labels) in distributions.drain() {
-            if let Some(desc) = descriptions.get(name.as_str()) {
-                write_help_line(&mut output, name.as_str(), desc);
-            }
+            if let Some(desc) = descriptions.get(name.as_str()) {}
+
+            let distribution_type = inner.distribution_builder.get_distribution_type(name.as_str());
 
-            let distribution_type = self.distribution_builder.get_distribution_type(name.as_str());
-            write_type_line(&mut output, name.as_str(), distribution_type);
             for (labels, distribution) in by_labels.drain(..) {
                 let (sum, count) = match distribution {
                     Distribution::Summary(summary, quantiles, sum) => {
                         let snapshot = summary.snapshot(Instant::now());
                         for quantile in quantiles.iter() {
                             let value = snapshot.quantile(quantile.value()).unwrap_or(0.0);
-                            write_metric_line(
-                                &mut output,
-                                &name,
-                                None,
-                                &labels,
-                                Some(("quantile", quantile.value())),
-                                value,
-                            );
                         }
 
                         (sum, summary.count() as u64)
                     }
                     Distribution::Histogram(histogram) => {
-                        for (le, count) in histogram.buckets() {
-                            write_metric_line(
-                                &mut output,
-                                &name,
-                                Some("bucket"),
-                                &labels,
-                                Some(("le", le)),
-                                count,
-                            );
-                        }
-                        write_metric_line(
-                            &mut output,
-                            &name,
-                            Some("bucket"),
-                            &labels,
-                            Some(("le", "+Inf")),
-                            histogram.count(),
-                        );
+                        for (le, count) in histogram.buckets() {}
 
                         (histogram.sum(), histogram.count())
                     }
                 };
-
-                write_metric_line::<&str, f64>(&mut output, &name, Some("sum"), &labels, None, sum);
-                write_metric_line::<&str, u64>(
-                    &mut output,
-                    &name,
-                    Some("count"),
-                    &labels,
-                    None,
-                    count,
-                );
             }
-
-            output.push('\n');
         }
         req
     }

From 29f759192521245024142d17c7fa21d827455ac4 Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Mon, 4 Nov 2024 19:57:58 +0800
Subject: [PATCH 10/11] 1

---
 .../src/remote_write_proto.rs                     | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/metrics-exporter-prometheus/src/remote_write_proto.rs b/metrics-exporter-prometheus/src/remote_write_proto.rs
index 1c4020e7..2e788728 100644
--- a/metrics-exporter-prometheus/src/remote_write_proto.rs
+++ b/metrics-exporter-prometheus/src/remote_write_proto.rs
@@ -68,18 +68,19 @@ impl WriteRequest {
         let mut req = WriteRequest::default();
         let mut all_series = std::collections::HashMap::<String, TimeSeries>::new();
         for (name, mut by_labels) in counters.drain() {
-            if let Some(desc) = descriptions.get(name.as_str()) {}
-
             for (labels, value) in by_labels.drain() {}
         }
         for (name, mut by_labels) in gauges.drain() {
-            if let Some(desc) = descriptions.get(name.as_str()) {}
-
-            for (labels, value) in by_labels.drain() {}
+            let mut labels = vec![];
+            labels.push(Label { name: LABEL_NAME.to_string(), value: name });
+            for (labels, value) in by_labels.drain() {
+                // labels.into_iter().map(|f|Label {
+                //     name:f,
+                //     value
+                // }).collect();
+            }
         }
         for (name, mut by_labels) in distributions.drain() {
-            if let Some(desc) = descriptions.get(name.as_str()) {}
-
             let distribution_type = inner.distribution_builder.get_distribution_type(name.as_str());
 
             for (labels, distribution) in by_labels.drain(..) {

From 146d110413ff61b9e0f34c471cd6aa8cebf180ea Mon Sep 17 00:00:00 2001
From: taikulawo <qaq1362211689@gmail.com>
Date: Mon, 4 Nov 2024 20:05:18 +0800
Subject: [PATCH 11/11] 1

---
 .../src/remote_write_proto.rs                 | 141 +++++++++---------
 1 file changed, 72 insertions(+), 69 deletions(-)

diff --git a/metrics-exporter-prometheus/src/remote_write_proto.rs b/metrics-exporter-prometheus/src/remote_write_proto.rs
index 2e788728..2a7c7588 100644
--- a/metrics-exporter-prometheus/src/remote_write_proto.rs
+++ b/metrics-exporter-prometheus/src/remote_write_proto.rs
@@ -103,74 +103,6 @@ impl WriteRequest {
         }
         req
     }
-    /// Parse metrics from the Prometheus text format, and convert them into a
-    /// [`WriteRequest`].
-    fn from_text_format(text: String) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
-        fn samples_to_timeseries(
-            samples: Vec<prometheus_parse::Sample>,
-        ) -> Result<Vec<TimeSeries>, Box<dyn std::error::Error + Send + Sync>> {
-            let mut all_series = std::collections::HashMap::<String, TimeSeries>::new();
-
-            for sample in &samples {
-                let mut labels =
-                    sample.labels.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect::<Vec<_>>();
-
-                labels.push((LABEL_NAME, sample.metric.as_str()));
-
-                labels.sort_by(|a, b| a.0.cmp(b.0));
-
-                let mut ident = sample.metric.clone();
-                ident.push_str("_$$_");
-                for (k, v) in &labels {
-                    ident.push_str(k);
-                    ident.push('=');
-                    ident.push_str(v);
-                }
-
-                let series = all_series.entry(ident).or_insert_with(|| {
-                    let labels = labels
-                        .iter()
-                        .map(|(k, v)| Label { name: k.to_string(), value: v.to_string() })
-                        .collect::<Vec<_>>();
-
-                    TimeSeries { labels, samples: vec![] }
-                });
-
-                let value = match sample.value {
-                    prometheus_parse::Value::Counter(v) => v,
-                    prometheus_parse::Value::Gauge(v) => v,
-                    prometheus_parse::Value::Histogram(_) => {
-                        Err("histogram not supported yet".to_string())?
-                    }
-                    prometheus_parse::Value::Summary(_) => {
-                        Err("summary not supported yet".to_string())?
-                    }
-                    prometheus_parse::Value::Untyped(v) => v,
-                };
-
-                series
-                    .samples
-                    .push(Sample { value, timestamp: sample.timestamp.timestamp_millis() });
-            }
-
-            Ok(all_series.into_values().collect())
-        }
-
-        let iter = text.trim().lines().map(|x| Ok(x.to_string()));
-        let parsed = prometheus_parse::Scrape::parse(iter)
-            .map_err(|err| format!("could not parse input as Prometheus text format: {err}"))?;
-
-        let mut series = samples_to_timeseries(parsed.samples)?;
-        series.sort_by(|a, b| {
-            let name_a = a.labels.iter().find(|x| x.name == LABEL_NAME).unwrap();
-            let name_b = b.labels.iter().find(|x| x.name == LABEL_NAME).unwrap();
-            name_a.value.cmp(&name_b.value)
-        });
-
-        let s = WriteRequest { timeseries: series };
-
-        Ok(s.sorted())
-    }
 
     /// Build a fully prepared HTTP request that an be sent to a remote write endpoint.
     pub fn build_http_request(
@@ -254,6 +186,77 @@ pub struct Sample {
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    /// Parse metrics from the Prometheus text format, and convert them into a
+    /// [`WriteRequest`].
+    fn from_text_format(
+        text: String,
+    ) -> Result<WriteRequest, Box<dyn std::error::Error + Send + Sync>> {
+        fn samples_to_timeseries(
+            samples: Vec<prometheus_parse::Sample>,
+        ) -> Result<Vec<TimeSeries>, Box<dyn std::error::Error + Send + Sync>> {
+            let mut all_series = std::collections::HashMap::<String, TimeSeries>::new();
+
+            for sample in &samples {
+                let mut labels =
+                    sample.labels.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect::<Vec<_>>();
+
+                labels.push((LABEL_NAME, sample.metric.as_str()));
+
+                labels.sort_by(|a, b| a.0.cmp(b.0));
+
+                let mut ident = sample.metric.clone();
+                ident.push_str("_$$_");
+                for (k, v) in &labels {
+                    ident.push_str(k);
+                    ident.push('=');
+                    ident.push_str(v);
+                }
+
+                let series = all_series.entry(ident).or_insert_with(|| {
+                    let labels = labels
+                        .iter()
+                        .map(|(k, v)| Label { name: k.to_string(), value: v.to_string() })
+                        .collect::<Vec<_>>();
+
+                    TimeSeries { labels, samples: vec![] }
+                });
+
+                let value = match sample.value {
+                    prometheus_parse::Value::Counter(v) => v,
+                    prometheus_parse::Value::Gauge(v) => v,
+                    prometheus_parse::Value::Histogram(_) => {
+                        Err("histogram not supported yet".to_string())?
+                    }
+                    prometheus_parse::Value::Summary(_) => {
+                        Err("summary not supported yet".to_string())?
+                    }
+                    prometheus_parse::Value::Untyped(v) => v,
+                };
+
+                series
+                    .samples
+                    .push(Sample { value, timestamp: sample.timestamp.timestamp_millis() });
+            }
+
+            Ok(all_series.into_values().collect())
+        }
+
+        let iter = text.trim().lines().map(|x| Ok(x.to_string()));
+        let parsed = prometheus_parse::Scrape::parse(iter)
+            .map_err(|err| format!("could not parse input as Prometheus text format: {err}"))?;
+
+        let mut series = samples_to_timeseries(parsed.samples)?;
+        series.sort_by(|a, b| {
+            let name_a = a.labels.iter().find(|x| x.name == LABEL_NAME).unwrap();
+            let name_b = b.labels.iter().find(|x| x.name == LABEL_NAME).unwrap();
+            name_a.value.cmp(&name_b.value)
+        });
+
+        let s = WriteRequest { timeseries: series };
+
+        Ok(s.sorted())
+    }
     #[test]
     fn test_name() {
         let input = r#"
@@ -267,7 +270,7 @@ alpha 10 1000
 http_requests_total{method="post",code="200"} 50 1000
     "#;
 
-        let req = WriteRequest::from_text_format(input.to_string()).unwrap();
+        let req = from_text_format(input.to_string()).unwrap();
 
         assert_eq!(
             req,