Observability Support (#424)

core/main/Cargo.toml

@@ -47,6 +47,7 @@ serde_json = "1.0"
 base64 = "0.13.0"
 sd-notify = { version = "0.4.1", optional = true }
 exitcode = "1.1.2"
+rand = "0.8"
 
 
 [build-dependencies]

core/main/src/firebolt/firebolt_gateway.rs

@@ -34,7 +34,7 @@ use serde::Serialize;
 
 use crate::{
     firebolt::firebolt_gatekeeper::FireboltGatekeeper,
-    service::apps::app_events::AppEvents,
+    service::{apps::app_events::AppEvents, telemetry_builder::TelemetryBuilder},
     state::{bootstrap_state::BootstrapState, session_state::Session},
 };
 
@@ -158,8 +158,17 @@ impl FireboltGateway {
          */
         let mut request_c = request.clone();
         request_c.method = FireboltOpenRpcMethod::name_with_lowercase_module(&request.method);
+
+        let metrics_timer = TelemetryBuilder::start_firebolt_metrics_timer(
+            &platform_state.get_client().get_extn_client(),
+            request_c.method.clone(),
+            request_c.ctx.app_id.clone(),
+        );
+
         tokio::spawn(async move {
-            match FireboltGatekeeper::gate(platform_state.clone(), request_c.clone()).await {
+            let result = FireboltGatekeeper::gate(platform_state.clone(), request_c.clone()).await;
+
+            match result {
                 Ok(_) => {
                     // Route
                     match request.clone().ctx.protocol {
@@ -182,7 +191,13 @@ impl FireboltGateway {
                                 .get_session(&request_c.ctx)
                             {
                                 // if the websocket disconnects before the session is recieved this leads to an error
-                                RpcRouter::route(platform_state, request_c, session).await;
+                                RpcRouter::route(
+                                    platform_state.clone(),
+                                    request_c,
+                                    session,
+                                    metrics_timer.clone(),
+                                )
+                                .await;
                             } else {
                                 error!("session is missing request is not forwarded");
                             }
@@ -191,11 +206,19 @@ impl FireboltGateway {
                 }
                 Err(e) => {
                     let deny_reason = e.reason;
-                    // return error for Api message
+
+                    TelemetryBuilder::stop_and_send_firebolt_metrics_timer(
+                        &platform_state.clone(),
+                        metrics_timer,
+                        format!("{}", deny_reason.get_observability_error_code()),
+                    )
+                    .await;
+
                     error!(
                         "Failed gateway present error {:?} {:?}",
                         request, deny_reason
                     );
+
                     let caps = e.caps.iter().map(|x| x.as_str()).collect();
                     let err = JsonRpcMessage {
                         jsonrpc: TwoPointZero {},
@@ -206,7 +229,9 @@ impl FireboltGateway {
                             data: None,
                         }),
                     };
+
                     let msg = serde_json::to_string(&err).unwrap();
+
                     let api_msg = ApiMessage::new(
                         request.clone().ctx.protocol,
                         msg,

core/main/src/firebolt/handlers/metrics_management_rpc.rs

@@ -92,7 +92,7 @@ impl MetricsManagementServer for MetricsManagementImpl {
         if let Some(device_session_id) = context_params.context.device_session_id {
             self.state
                 .metrics
-                .update_session_id(Some(device_session_id));
+                .update_session_id(self.state.clone(), Some(device_session_id));
         }
         Ok(())
     }
@@ -105,9 +105,10 @@ impl MetricsManagementServer for MetricsManagementImpl {
         for key in request.keys {
             // currently handling only one key which is deviceSessionId
             if key.as_str() == "deviceSessionId" {
-                self.state
-                    .metrics
-                    .update_session_id(Some(String::from(&self.state.device_session_id)));
+                self.state.metrics.update_session_id(
+                    self.state.clone(),
+                    Some(String::from(&self.state.device_session_id)),
+                );
             }
         }
         Ok(())

core/main/src/firebolt/rpc_router.rs

@@ -30,6 +30,7 @@ use jsonrpsee::{
 use ripple_sdk::{
     api::{
         apps::EffectiveTransport,
+        firebolt::fb_metrics::Timer,
         gateway::rpc_gateway_api::{ApiMessage, JsonRpcApiResponse, RpcRequest},
     },
     chrono::Utc,
@@ -132,15 +133,36 @@ async fn resolve_route(
 }
 
 impl RpcRouter {
-    pub async fn route(state: PlatformState, req: RpcRequest, session: Session) {
+    pub async fn route(
+        state: PlatformState,
+        req: RpcRequest,
+        session: Session,
+        timer: Option<Timer>,
+    ) {
         let methods = state.router_state.get_methods();
         let resources = state.router_state.resources.clone();
+
         tokio::spawn(async move {
             let method = req.method.clone();
             let app_id = req.ctx.app_id.clone();
             let session_id = req.ctx.session_id.clone();
             let start = Utc::now().timestamp_millis();
-            if let Ok(msg) = resolve_route(methods, resources, req.clone()).await {
+            let resp = resolve_route(methods, resources, req.clone()).await;
+
+            let status = match resp.clone() {
+                Ok(msg) => {
+                    if msg.is_error() {
+                        msg.jsonrpc_msg
+                    } else {
+                        "0".into()
+                    }
+                }
+                Err(e) => format!("{}", e),
+            };
+
+            TelemetryBuilder::stop_and_send_firebolt_metrics_timer(&state, timer, status).await;
+
+            if let Ok(msg) = resp {
                 let now = Utc::now().timestamp_millis();
                 let success = !msg.is_error();
                 info!(
@@ -154,7 +176,9 @@ impl RpcRouter {
                         false => &msg.jsonrpc_msg,
                     }
                 );
+
                 TelemetryBuilder::send_fb_tt(&state, req.clone(), now - start, success);
+
                 match session.get_transport() {
                     EffectiveTransport::Websocket => {
                         if let Err(e) = session.send_json_rpc(msg).await {

core/main/src/processor/metrics_processor.rs

@@ -199,9 +199,10 @@ impl ExtnRequestProcessor for MetricsProcessor {
                     Err(e) => Self::handle_error(client, msg, e).await,
                 }
             }
-            MetricsPayload::OperationalMetric(t) => {
+            MetricsPayload::TelemetryPayload(t) => {
                 TelemetryBuilder::update_session_id_and_send_telemetry(&state, t).is_ok()
             }
+            MetricsPayload::OperationalMetric(_) => true,
         }
     }
 }
@@ -257,6 +258,7 @@ impl ExtnRequestProcessor for OpMetricsProcessor {
             OperationalMetricRequest::UnSubscribe => state
                 .metrics
                 .operational_telemetry_listener(&requestor, true),
+            _ => (),
         }
         Self::ack(state.get_client().get_extn_client(), msg)
             .await

core/main/src/service/observability/mod.rs

@@ -0,0 +1,11 @@
+use std::sync::Arc;
+
+use crate::state::platform_state::PlatformState;
+use ripple_sdk::api::firebolt::fb_telemetry::OperationalMetricRequest;
+static mut PLATFORM_STATE: Option<Arc<PlatformState>> = None;
+pub struct ObservabilityClient {}
+impl ObservabilityClient {
+    pub fn report(platform_state: &PlatformState, payload: OperationalMetricRequest) {
+        println!("payload: {:?}", payload);
+    }
+}

core/main/src/service/telemetry_builder.rs

@@ -18,7 +18,10 @@
 use ripple_sdk::{
     api::{
         firebolt::{
-            fb_metrics::{ErrorParams, InternalInitializeParams, SystemErrorParams},
+            fb_metrics::{
+                get_metrics_tags, ErrorParams, InteractionType, InternalInitializeParams,
+                SystemErrorParams, Tag, Timer,
+            },
             fb_telemetry::{
                 AppLoadStart, AppLoadStop, FireboltInteraction, InternalInitialize,
                 TelemetryAppError, TelemetryPayload, TelemetrySignIn, TelemetrySignOut,
@@ -28,8 +31,9 @@ use ripple_sdk::{
         gateway::rpc_gateway_api::{CallContext, RpcRequest},
     },
     chrono::{DateTime, Utc},
+    extn::client::extn_client::ExtnClient,
     framework::RippleResponse,
-    log::error,
+    log::{debug, error},
 };
 use serde_json::Value;
 
@@ -199,4 +203,47 @@ impl TelemetryBuilder {
             error!("send_telemetry={:?}", e)
         }
     }
+
+    pub fn start_firebolt_metrics_timer(
+        extn_client: &ExtnClient,
+        name: String,
+        app_id: String,
+    ) -> Option<Timer> {
+        let metrics_context = extn_client.get_metrics_context();
+
+        if !metrics_context.enabled {
+            return None;
+        }
+
+        let metrics_tags = get_metrics_tags(extn_client, InteractionType::Firebolt, Some(app_id));
+
+        debug!("start_firebolt_metrics_timer: {}: {:?}", name, metrics_tags);
+
+        Some(Timer::start(name, Some(metrics_tags)))
+    }
+
+    pub async fn stop_and_send_firebolt_metrics_timer(
+        ps: &PlatformState,
+        timer: Option<Timer>,
+        status: String,
+    ) {
+        if let Some(mut timer) = timer {
+            timer.stop();
+            timer.insert_tag(Tag::Status.key(), status);
+            if let Err(e) = &ps
+                .get_client()
+                .send_extn_request(
+                    ripple_sdk::api::firebolt::fb_telemetry::OperationalMetricRequest::Timer(
+                        timer.clone(),
+                    ),
+                )
+                .await
+            {
+                error!(
+                    "stop_and_send_firebolt_metrics_timer: send_telemetry={:?}",
+                    e
+                )
+            }
+        }
+    }
 }

core/main/src/state/cap/permitted_state.rs

@@ -23,6 +23,7 @@ use std::{
 
 use ripple_sdk::{
     api::{
+        config::FEATURE_CLOUD_PERMISISONS,
         device::device_apps::AppsRequest,
         //config::Config,
         distributor::distributor_permissions::{PermissionRequest, PermissionResponse},
@@ -149,7 +150,12 @@ impl PermissionHandler {
         app_id: &str,
         allow_cached: bool,
     ) -> RippleResponse {
-        if state.get_device_manifest().get_features().cloud_permissions {
+        if state
+            .get_client()
+            .get_extn_client()
+            .get_features()
+            .contains(&String::from(FEATURE_CLOUD_PERMISISONS))
+        {
             if allow_cached {
                 if let Some(permissions) =
                     state.cap_state.permitted_state.get_app_permissions(app_id)