Merge pull request #7 from schultyy/metrics

Add service metrics

Author: schultyy

Cargo.toml

@@ -12,12 +12,13 @@ tracing-subscriber = { version = "0.3.19", features = ["env-filter", "json"] }
 clap = { version = "4.4", features = ["derive"] }
 fake = { version = "4", features = ["derive"] }
 tracing-opentelemetry = "0.29.0"
-opentelemetry = "0.29.0"
+opentelemetry = { version = "0.29.0", features = ["metrics"] }
 opentelemetry-otlp = { version = "0.29.0", features = ["tonic", "grpc-tonic"] }
 opentelemetry_sdk = { version = "0.29.0", features = [
   "rt-tokio",
   "tokio",
   "opentelemetry-http",
+  "metrics",
 ] }
 tonic = "0.12.3"
 opentelemetry-appender-tracing = "0.29.0"

src/code_gen/instruction.rs

@@ -85,6 +85,32 @@ pub const CHECK_INTERRUPT_CODE: u8 = 0x12;
 pub const CALL_CODE: u8 = 0x13;
 pub const RET_CODE: u8 = 0x14;
 
+pub fn code_to_name(code: u8) -> String {
+    match code {
+        PUSH_STRING_CODE => "PushString".to_string(),
+        PUSH_INT_CODE => "PushInt".to_string(),
+        POP_CODE => "Pop".to_string(),
+        DEC_CODE => "Dec".to_string(),
+        JMP_IF_ZERO_CODE => "JmpIfZero".to_string(),
+        LABEL_CODE => "Label".to_string(),
+        STDOUT_CODE => "Stdout".to_string(),
+        STDERR_CODE => "Stderr".to_string(),
+        SLEEP_CODE => "Sleep".to_string(),
+        STORE_VAR_CODE => "StoreVar".to_string(),
+        LOAD_VAR_CODE => "LoadVar".to_string(),
+        DUP_CODE => "Dup".to_string(),
+        JUMP_CODE => "Jump".to_string(),
+        PRINTF_CODE => "Printf".to_string(),
+        REMOTE_CALL_CODE => "RemoteCall".to_string(),
+        START_CONTEXT_CODE => "StartContext".to_string(),
+        END_CONTEXT_CODE => "EndContext".to_string(),
+        CHECK_INTERRUPT_CODE => "CheckInterrupt".to_string(),
+        CALL_CODE => "Call".to_string(),
+        RET_CODE => "Ret".to_string(),
+        _ => "Unknown".to_string(),
+    }
+}
+
 impl Instruction {
     pub fn code(&self) -> u8 {
         match self {

src/main.rs

@@ -130,11 +130,14 @@ async fn execute_service(
     let tracer = vm::setup_tracer(&otel_endpoint, &service_name)
         .map_err(|e| RuntimeError::InitTraceError(e))?;
 
+    let meter_provider = vm::init_meter_provider(Some(&otel_endpoint), &service_name)
+        .map_err(|e| RuntimeError::InitMeterError(e))?;
+
     let mut vm = vm::VM::new(service_code.clone(), &service_name, print_tx)
         .with_remote_call_tx(coordinator.get_main_tx().clone())
         .with_remote_call_rx(remote_call_rx)
-        .with_tracer(tracer.clone());
-
+        .with_tracer(tracer.clone())
+        .with_meter_provider(meter_provider);
     if let Some(remote_call_limit) = args.remote_call_limit {
         vm = vm.with_custom_remote_call_limit(remote_call_limit);
     }

src/runtime_error.rs

@@ -7,6 +7,7 @@ pub enum RuntimeError {
     VMError(vm::VMError),
     ServiceError(JoinError),
     InitTraceError(opentelemetry_otlp::ExporterBuildError),
+    InitMeterError(opentelemetry_otlp::ExporterBuildError),
 }
 
 impl std::error::Error for RuntimeError {}
@@ -17,6 +18,7 @@ impl std::fmt::Display for RuntimeError {
             RuntimeError::VMError(e) => write!(f, "VM error: {}", e),
             RuntimeError::ServiceError(e) => write!(f, "Service error: {}", e),
             RuntimeError::InitTraceError(e) => write!(f, "Init trace error: {}", e),
+            RuntimeError::InitMeterError(e) => write!(f, "Init meter error: {}", e),
         }
     }
 }

src/vm.rs

@@ -1,5 +1,8 @@
 use std::collections::HashMap;
 
+use opentelemetry::metrics::Counter;
+use opentelemetry::metrics::Gauge;
+use opentelemetry::metrics::MeterProvider;
 use opentelemetry::propagation::TextMapPropagator;
 use opentelemetry::trace::{TraceContextExt, TracerProvider};
 use opentelemetry::{global, KeyValue};
@@ -8,6 +11,8 @@ use opentelemetry::{
     Context,
 };
 use opentelemetry_otlp::{WithExportConfig, WithTonicConfig};
+use opentelemetry_sdk::metrics::SdkMeterProvider;
+use opentelemetry_sdk::metrics::Temporality;
 use opentelemetry_sdk::propagation::TraceContextPropagator;
 use opentelemetry_sdk::trace::SdkTracerProvider;
 use opentelemetry_sdk::Resource;
@@ -105,6 +110,40 @@ pub fn setup_tracer(
     Ok(provider)
 }
 
+pub(crate) fn init_meter_provider(
+    endpoint: Option<&str>,
+    service_name: &str,
+) -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, opentelemetry_otlp::ExporterBuildError> {
+    let provider = if let Some(endpoint) = endpoint {
+        let exporter = opentelemetry_otlp::MetricExporter::builder()
+            .with_temporality(Temporality::Delta)
+            .with_tonic()
+            .with_endpoint(endpoint.to_string())
+            .build()?;
+        let resource = Resource::builder()
+            .with_service_name(service_name.to_string())
+            .build();
+
+        SdkMeterProvider::builder()
+            .with_periodic_exporter(exporter)
+            .with_resource(resource)
+            .build()
+    } else {
+        let exporter = opentelemetry_stdout::MetricExporter::default();
+
+        let resource = Resource::builder()
+            .with_service_name(service_name.to_string())
+            .build();
+
+        SdkMeterProvider::builder()
+            .with_periodic_exporter(exporter)
+            .with_resource(resource)
+            .build()
+    };
+
+    Ok(provider)
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum PrintMessage {
     Stdout(String),
@@ -130,6 +169,7 @@ pub struct VM {
     remote_call_limit: usize,
     service_name: String,
     tracer: Option<SdkTracerProvider>,
+    meter_provider: SdkMeterProvider,
     otel_context: Option<opentelemetry::Context>,
 }
 
@@ -161,7 +201,9 @@ impl VM {
         service_name: &str,
         print_tx: mpsc::Sender<PrintMessage>,
     ) -> Self {
+        let service_name = service_name.to_string();
         let (code, label_jump_map, label_index_map) = generate_bytecode(code);
+
         Self {
             code,
             label_jump_map,
@@ -179,6 +221,7 @@ impl VM {
             service_name: service_name.to_string(),
             tracer: None,
             otel_context: None,
+            meter_provider: init_meter_provider(None, &service_name).unwrap(),
         }
     }
 
@@ -207,13 +250,63 @@ impl VM {
         self
     }
 
+    pub fn with_meter_provider(mut self, meter_provider: SdkMeterProvider) -> Self {
+        self.meter_provider = meter_provider;
+        self
+    }
+
+    fn build_counters(
+        &self,
+    ) -> Result<(Counter<u64>, Counter<u64>, Gauge<u64>, Gauge<u64>), VMError> {
+        let remote_invocation_counter = self
+            .meter_provider
+            .meter("remote_invocation_counter")
+            .u64_counter("remote_invocation_counter")
+            .build()
+            .to_owned();
+
+        let local_invocation_counter = self
+            .meter_provider
+            .meter("local_invocation_counter")
+            .u64_counter("local_invocation_counter")
+            .build()
+            .to_owned();
+
+        let instruction_duration = self
+            .meter_provider
+            .meter("instruction_duration")
+            .u64_gauge("instruction_duration")
+            .with_unit("ms")
+            .with_description("The duration of executing an instruction in milliseconds")
+            .build()
+            .to_owned();
+
+        let remote_call_duration = self
+            .meter_provider
+            .meter("remote_call_duration")
+            .u64_gauge("remote_call_duration")
+            .with_unit("ms")
+            .with_description("The duration of a remote call in milliseconds")
+            .build()
+            .to_owned();
+
+        Ok((
+            remote_invocation_counter,
+            local_invocation_counter,
+            instruction_duration,
+            remote_call_duration,
+        ))
+    }
+
     pub async fn run(&mut self) -> Result<(), VMError> {
         let mut execution_counter = 0;
+        let counters = self.build_counters()?;
+
         while self.ip < self.code.len() {
             if self.ip >= self.code.len() {
                 return Err(VMError::IPOutOfBounds(self.ip, self.code.len()));
             }
-            self.execute_instruction().await?;
+            self.execute_instruction(counters.clone()).await?;
             execution_counter += 1;
             if let Some(max_execution_counter) = self.max_execution_counter {
                 if execution_counter > max_execution_counter {
@@ -261,8 +354,18 @@ impl VM {
         self.stack.last_mut().ok_or(VMError::MissingStackFrame)
     }
 
-    async fn execute_instruction(&mut self) -> Result<(), VMError> {
+    async fn execute_instruction(
+        &mut self,
+        counters: (Counter<u64>, Counter<u64>, Gauge<u64>, Gauge<u64>),
+    ) -> Result<(), VMError> {
         let instruction = self.code[self.ip];
+        let (
+            remote_invocation_counter,
+            local_invocation_counter,
+            instruction_duration,
+            remote_call_duration,
+        ) = counters;
+        let start = std::time::Instant::now();
         match instruction {
             PUSH_STRING_CODE => {
                 let (_start, end, str_len) = self.extract_length();
@@ -443,6 +546,7 @@ impl VM {
                 self.ip += 1;
             }
             REMOTE_CALL_CODE => {
+                let start = std::time::Instant::now();
                 let remote_call_tx = self
                     .remote_call_tx
                     .as_ref()
@@ -497,6 +601,23 @@ impl VM {
                     .await
                     .map_err(|e| VMError::RemoteCallError(e.to_string()))?;
 
+                remote_invocation_counter.add(
+                    1,
+                    &[
+                        KeyValue::new("service", self.service_name.clone()),
+                        KeyValue::new("method", remote_method.to_string().clone()),
+                    ],
+                );
+
+                let duration = start.elapsed();
+                let duration_ms = duration.as_millis() as u64;
+                remote_call_duration.record(
+                    duration_ms,
+                    &[
+                        KeyValue::new("service", self.service_name.clone()),
+                        KeyValue::new("method", remote_method.to_string().clone()),
+                    ],
+                );
                 if let Some(cx) = cx {
                     cx.span()
                         .set_attributes(vec![KeyValue::new("response", "OK")]);
@@ -537,7 +658,9 @@ impl VM {
                 let (_start, end, label_len) = self.extract_length();
                 let label = &self.code[end..end + label_len];
                 let label = String::from_utf8(label.to_vec()).unwrap();
-                self.handle_local_call(label).await?;
+                self.handle_local_call(label.clone()).await?;
+                local_invocation_counter
+                    .add(1, &[KeyValue::new("method", label.to_string().clone())]);
             }
             RET_CODE => {
                 self.ip = self.return_addresses.pop().unwrap();
@@ -547,6 +670,15 @@ impl VM {
                 return Err(VMError::InvalidInstruction(instruction));
             }
         }
+        let duration = start.elapsed();
+        let duration_ms = duration.as_millis() as u64;
+        instruction_duration.record(
+            duration_ms,
+            &[KeyValue::new(
+                "instruction",
+                crate::code_gen::instruction::code_to_name(instruction),
+            )],
+        );
         Ok(())
     }