Add prometheus parse and scrape

prometheus parse will parse prometheus or openmetrics format files
provided as pipeline input and output parsed metrics data

prometheus scrape will scrape a prometheus target and output parsed
metrics data

Author: drbrain

Cargo.lock

@@ -18,6 +18,7 @@ categories = ["command-line-utilities"]
 chrono = "0.4.39"
 nom = "8"
 nom-language = "0.1"
+nom-openmetrics = { git = "https://github.com/drbrain/nom-openmetrics" }
 nu-plugin = "0.102.0"
 nu-protocol = { version = "0.102.0", features = [ "plugin" ] }
 prometheus-http-query = "0.8.3"

Cargo.toml

@@ -3,9 +3,11 @@ mod label_names_builder;
 mod label_values;
 mod label_values_builder;
 mod metric_metadata;
+mod parse;
 mod query_builder;
 mod query_instant;
 mod query_range;
+mod scrape;
 mod selector_parser;
 mod series;
 mod targets;
@@ -16,9 +18,12 @@ pub use label_values::LabelValues;
 pub use label_values_builder::LabelValuesBuilder;
 pub use metric_metadata::MetricMetadata;
 use nu_protocol::{LabeledError, Span};
+pub use parse::Parse;
+pub use parse::ParseFormat;
 pub use query_builder::QueryBuilder;
 pub use query_instant::QueryInstant;
 pub use query_range::QueryRange;
+pub use scrape::Scrape;
 pub use selector_parser::SelectorParser;
 pub use series::Series;
 pub use targets::Targets;

src/client.rs

@@ -0,0 +1,101 @@
+use crate::Client;
+use nom_openmetrics::{
+    parser::{exposition, prometheus},
+    Family, MetricDescriptor, Sample,
+};
+use nu_protocol::{record, LabeledError, Record, Span, Value};
+
+#[derive(Default)]
+pub enum ParseFormat {
+    #[default]
+    Prometheus,
+    Openmetrics,
+}
+
+pub struct Parse<'a> {
+    input: &'a Value,
+    format: ParseFormat,
+}
+
+impl<'a> Parse<'a> {
+    pub fn new(input: &'a Value) -> Self {
+        Self {
+            input,
+            format: Default::default(),
+        }
+    }
+
+    pub fn run(self) -> Result<Value, LabeledError> {
+        let Self { input, format } = self;
+
+        let (_, families) = match format {
+            ParseFormat::Prometheus => prometheus(input.as_str()?).unwrap(),
+            ParseFormat::Openmetrics => exposition(input.as_str()?).unwrap(),
+        };
+
+        let families = families
+            .iter()
+            .map(|family| family_to_value(family))
+            .collect();
+
+        Ok(Value::list(families, Span::unknown()))
+    }
+
+    pub fn set_format(&mut self, format: ParseFormat) {
+        self.format = format;
+    }
+}
+
+impl<'a> Client for Parse<'a> {}
+
+fn family_to_value(family: &Family) -> Value {
+    let descriptors = family
+        .descriptors
+        .iter()
+        .map(|descriptor| descriptor_to_value(descriptor))
+        .collect();
+
+    let samples = family
+        .samples
+        .iter()
+        .map(|sample| sample_to_value(sample))
+        .collect();
+
+    let record = record! {
+        "descriptors" => Value::list(descriptors, Span::unknown()),
+        "samples" => Value::list(samples, Span::unknown()),
+    };
+
+    Value::record(record, Span::unknown())
+}
+
+fn descriptor_to_value(descriptor: &MetricDescriptor) -> Value {
+    let record = match descriptor {
+        MetricDescriptor::Type { metric, r#type } => record! {
+            "descriptor" => Value::string("type", Span::unknown()),
+            "metric" => Value::string(*metric, Span::unknown()),
+            "type" => Value::string(r#type.to_string(), Span::unknown())
+        },
+        MetricDescriptor::Help { metric, help } => record! {
+            "descriptor" => Value::string("help", Span::unknown()),
+            "metric" => Value::string(*metric, Span::unknown()),
+            "help" => Value::string(help, Span::unknown())
+        },
+        MetricDescriptor::Unit { metric, unit } => record! {
+            "descriptor" => Value::string("unit", Span::unknown()),
+            "metric" => Value::string(*metric, Span::unknown()),
+            "unit" => Value::string(*unit, Span::unknown())
+        },
+    };
+
+    Value::record(record, Span::unknown())
+}
+
+fn sample_to_value(sample: &Sample) -> Value {
+    let mut record = Record::new();
+
+    record.insert("name", Value::string(sample.name(), Span::unknown()));
+    record.insert("value", Value::float(sample.number(), Span::unknown()));
+
+    Value::record(record, Span::unknown())
+}

src/client/parse.rs

@@ -0,0 +1,58 @@
+use nom_openmetrics::{Family, Sample};
+use nu_protocol::{LabeledError, Record, Span, Value};
+
+use crate::client::Client;
+
+pub struct Scrape {
+    target: String,
+}
+
+impl Scrape {
+    pub fn new(target: String) -> Self {
+        Self { target }
+    }
+
+    pub fn run(self) -> Result<Value, LabeledError> {
+        let Self { ref target, .. } = self;
+
+        self.runtime()?.block_on(async {
+            let body = reqwest::get(target).await.unwrap().bytes().await.unwrap();
+
+            let body = String::from_utf8(body.to_vec()).unwrap();
+
+            let (_, families) = nom_openmetrics::parser::prometheus(&body).unwrap();
+
+            let families = families
+                .iter()
+                .map(|family| family_to_value(family))
+                .collect();
+
+            Ok(Value::list(families, Span::unknown()))
+        })
+    }
+}
+
+impl Client for Scrape {}
+
+fn family_to_value(family: &Family) -> Value {
+    let mut record = Record::new();
+
+    let samples = family
+        .samples
+        .iter()
+        .map(|sample| sample_to_value(sample))
+        .collect();
+
+    record.insert("samples", Value::list(samples, Span::unknown()));
+
+    Value::record(record, Span::unknown())
+}
+
+fn sample_to_value(sample: &Sample) -> Value {
+    let mut record = Record::new();
+
+    record.insert("name", Value::string(sample.name(), Span::unknown()));
+    record.insert("value", Value::float(sample.number(), Span::unknown()));
+
+    Value::record(record, Span::unknown())
+}

src/client/scrape.rs

@@ -4,11 +4,11 @@ mod query;
 mod source;
 
 use client::Client;
-use nu_plugin::{serve_plugin, JsonSerializer};
+use nu_plugin::{serve_plugin, MsgPackSerializer};
 use prometheus::Prometheus;
 use query::Query;
 use source::Source;
 
 fn main() {
-    serve_plugin(&Prometheus, JsonSerializer)
+    serve_plugin(&Prometheus, MsgPackSerializer)
 }

src/main.rs

@@ -1,9 +1,11 @@
 mod label_names_command;
 mod label_values_command;
 mod metric_metadata_command;
+mod parse_command;
 mod prometheus_command;
 mod query_command;
 mod query_range_command;
+mod scrape_command;
 mod series_command;
 mod sources_command;
 mod targets_command;
@@ -16,6 +18,8 @@ use crate::prometheus::{
     targets_command::TargetsCommand,
 };
 use nu_plugin::Plugin;
+use parse_command::ParseCommand;
+use scrape_command::ScrapeCommand;
 
 #[derive(Clone)]
 pub struct Prometheus;
@@ -26,10 +30,12 @@ impl Plugin for Prometheus {
             Box::new(LabelNamesCommand),
             Box::new(LabelValuesCommand),
             Box::new(MetricMetadataCommand),
+            Box::new(ParseCommand),
             Box::new(PrometheusCommand),
             Box::new(QueryCommand),
             Box::new(QueryRangeCommand),
             Box::new(SeriesCommand),
+            Box::new(ScrapeCommand),
             Box::new(SourcesCommand),
             Box::new(TargetsCommand),
         ]

src/prometheus.rs

@@ -0,0 +1,60 @@
+use crate::{
+    client::{Parse, ParseFormat},
+    Prometheus,
+};
+use nu_plugin::{EngineInterface, EvaluatedCall, SimplePluginCommand};
+use nu_protocol::{LabeledError, Signature, Span, SyntaxShape, Type, Value};
+
+#[derive(Clone, Default)]
+pub struct ParseCommand;
+
+impl SimplePluginCommand for ParseCommand {
+    type Plugin = Prometheus;
+
+    fn name(&self) -> &str {
+        "prometheus parse"
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::build(self.name())
+            .description(self.description())
+            .named(
+                "format",
+                SyntaxShape::String,
+                "Metrics format, prometheus (default) or openmetrics",
+                None,
+            )
+            .input_output_types(vec![(Type::String, Type::table())])
+    }
+
+    fn description(&self) -> &str {
+        "Parse prometheus or openmetrics output"
+    }
+
+    fn run(
+        &self,
+        _plugin: &Self::Plugin,
+        _engine: &EngineInterface,
+        call: &EvaluatedCall,
+        input: &Value,
+    ) -> Result<Value, LabeledError> {
+        let format = call
+            .get_flag_value("format")
+            .unwrap_or(Value::string("prometheus", Span::unknown()));
+        let format_span = format.span();
+        let format = format.into_string()?;
+
+        let mut parser = Parse::new(input);
+
+        match format.as_str() {
+            "prometheus" => parser.set_format(ParseFormat::Prometheus),
+            "openmetrics" => parser.set_format(ParseFormat::Openmetrics),
+            _ => {
+                return Err(LabeledError::new("Invalid format")
+                    .with_label("must be prometheus or openmetrics", format_span));
+            }
+        }
+
+        parser.run()
+    }
+}