Feature/data governance cleanup

core/main/src/bootstrap/extn/load_session_step.rs

@@ -16,13 +16,10 @@
 //
 
 use ripple_sdk::framework::bootstrap::Bootstep;
-use ripple_sdk::tokio;
 use ripple_sdk::{async_trait::async_trait, framework::RippleResponse};
 
 use crate::processor::main_context_processor::MainContextProcessor;
-use crate::service::context_manager::ContextManager;
 use crate::state::bootstrap_state::BootstrapState;
-use crate::state::metrics_state::MetricsState;
 
 pub struct LoadDistributorValuesStep;
 
@@ -33,14 +30,8 @@ impl Bootstep<BootstrapState> for LoadDistributorValuesStep {
     }
 
     async fn setup(&self, s: BootstrapState) -> RippleResponse {
-        let mut ps = s.platform_state.clone();
-        tokio::spawn(async move {
-            MetricsState::initialize(&mut ps).await;
-        });
-
         MainContextProcessor::remove_expired_and_inactive_entries(&s.platform_state);
 
-        ContextManager::setup(&s.platform_state).await;
         if !s.platform_state.supports_session() {
             return Ok(());
         }

core/main/src/bootstrap/setup_extn_client_step.rs

@@ -19,8 +19,7 @@ use ripple_sdk::{
     async_trait::async_trait, framework::bootstrap::Bootstep, utils::error::RippleError,
 };
 
-use crate::processor::account_link_processor::AccountLinkProcessor;
-use crate::processor::metrics_processor::{MetricsProcessor, OpMetricsProcessor};
+use crate::processor::metrics_processor::OpMetricsProcessor;
 use crate::processor::settings_processor::SettingsProcessor;
 use crate::processor::{
     store_privacy_settings_processor::StorePrivacySettingsProcessor,
@@ -60,9 +59,7 @@ impl Bootstep<BootstrapState> for SetupExtnClientStep {
             state.platform_state.clone(),
         ));
         client.add_request_processor(AuthorizedInfoProcessor::new(state.platform_state.clone()));
-        client.add_request_processor(AccountLinkProcessor::new(state.platform_state.clone()));
         client.add_request_processor(SettingsProcessor::new(state.platform_state.clone()));
-        client.add_request_processor(MetricsProcessor::new(state.platform_state.clone()));
         client.add_request_processor(OpMetricsProcessor::new(state.platform_state.clone()));
         Ok(())
     }

core/main/src/bootstrap/start_fbgateway_step.rs

@@ -26,10 +26,9 @@ use crate::{
             audio_description_rpc::AudioDescriptionRPCProvider,
             authentication_rpc::AuthRPCProvider, capabilities_rpc::CapRPCProvider,
             closed_captions_rpc::ClosedcaptionsRPCProvider, device_rpc::DeviceRPCProvider,
-            discovery_rpc::DiscoveryRPCProvider, keyboard_rpc::KeyboardRPCProvider,
-            lcm_rpc::LifecycleManagementProvider, lifecycle_rpc::LifecycleRippleProvider,
-            localization_rpc::LocalizationRPCProvider,
-            metrics_management_rpc::MetricsManagementProvider, metrics_rpc::MetricsRPCProvider,
+            discovery_rpc::DiscoveryRPCProvider, internal_rpc::InternalProvider,
+            keyboard_rpc::KeyboardRPCProvider, lcm_rpc::LifecycleManagementProvider,
+            lifecycle_rpc::LifecycleRippleProvider, localization_rpc::LocalizationRPCProvider,
             parameters_rpc::ParametersRPCProvider, privacy_rpc::PrivacyProvider,
             profile_rpc::ProfileRPCProvider, provider_registrar::ProviderRegistrar,
             second_screen_rpc::SecondScreenRPCProvider,
@@ -70,14 +69,13 @@ impl FireboltGatewayStep {
         let _ = methods.merge(ParametersRPCProvider::provide_with_alias(state.clone()));
         let _ = methods.merge(SecureStorageRPCProvider::provide_with_alias(state.clone()));
         let _ = methods.merge(AdvertisingRPCProvider::provide_with_alias(state.clone()));
-        let _ = methods.merge(MetricsRPCProvider::provide_with_alias(state.clone()));
         let _ = methods.merge(DiscoveryRPCProvider::provide_with_alias(state.clone()));
         let _ = methods.merge(AuthRPCProvider::provide_with_alias(state.clone()));
         let _ = methods.merge(AccountRPCProvider::provide_with_alias(state.clone()));
-        let _ = methods.merge(MetricsManagementProvider::provide_with_alias(state.clone()));
         let _ = methods.merge(AudioDescriptionRPCProvider::provide_with_alias(
             state.clone(),
         ));
+        let _ = methods.merge(InternalProvider::provide_with_alias(state.clone()));
 
         // LCM Api(s) not required for internal launcher
         if !state.has_internal_launcher() {

core/main/src/broker/broker_utils.rs

@@ -72,6 +72,17 @@ impl BrokerUtils {
         }
     }
 
+    pub async fn process_for_app_main_request(
+        state: &mut PlatformState,
+        method: &str,
+        params: Option<Value>,
+        app_id: &str,
+    ) -> RpcResult<Value> {
+        let mut rpc_request = RpcRequest::internal(method, None).with_params(params);
+        rpc_request.ctx.app_id = app_id.to_owned();
+        Self::internal_request(state, rpc_request).await
+    }
+
     pub async fn process_internal_main_request<'a>(
         state: &mut PlatformState,
         method: &'a str,
@@ -90,7 +101,14 @@ impl BrokerUtils {
         state
             .metrics
             .add_api_stats(&rpc_request.ctx.request_id, method);
+        Self::internal_request(state, rpc_request).await
+    }
 
+    async fn internal_request(
+        state: &mut PlatformState,
+        rpc_request: RpcRequest,
+    ) -> RpcResult<Value> {
+        let method = rpc_request.method.clone();
         match state.internal_rpc_request(&rpc_request).await {
             Ok(res) => match res.as_value() {
                 Some(v) => Ok(v),

core/main/src/broker/endpoint_broker.rs

@@ -49,7 +49,7 @@ use crate::{
     broker::broker_utils::BrokerUtils,
     firebolt::firebolt_gateway::{FireboltGatewayCommand, JsonRpcError},
     service::extn::ripple_client::RippleClient,
-    state::{metrics_state::MetricsState, platform_state::PlatformState, session_state::Session},
+    state::{otel_state::OpMetricState, platform_state::PlatformState, session_state::Session},
     utils::router_utils::{
         add_telemetry_status_code, capture_stage, get_rpc_header, return_extn_response,
     },
@@ -372,7 +372,7 @@ pub struct EndpointBrokerState {
     cleaner_list: Arc<RwLock<Vec<BrokerCleaner>>>,
     reconnect_tx: Sender<BrokerConnectRequest>,
     provider_broker_state: ProvideBrokerState,
-    metrics_state: MetricsState,
+    metrics_state: OpMetricState,
 }
 impl Default for EndpointBrokerState {
     fn default() -> Self {
@@ -385,14 +385,14 @@ impl Default for EndpointBrokerState {
             cleaner_list: Arc::new(RwLock::new(Vec::new())),
             reconnect_tx: mpsc::channel(2).0,
             provider_broker_state: ProvideBrokerState::default(),
-            metrics_state: MetricsState::default(),
+            metrics_state: OpMetricState::default(),
         }
     }
 }
 
 impl EndpointBrokerState {
     pub fn new(
-        metrics_state: MetricsState,
+        metrics_state: OpMetricState,
         tx: Sender<BrokerOutput>,
         rule_engine: RuleEngine,
         ripple_client: RippleClient,
@@ -843,6 +843,10 @@ impl EndpointBrokerState {
         }
     }
 
+    pub fn has_rule(&self, method: &str) -> bool {
+        self.rule_engine.has_rule(method)
+    }
+
     pub fn get_rule(&self, rpc_request: &RpcRequest) -> Option<Rule> {
         self.rule_engine.get_rule(rpc_request)
     }
@@ -1607,7 +1611,7 @@ mod tests {
                 endpoint_broker::tests::RippleClient,
                 rules_engine::{Rule, RuleEngine, RuleSet, RuleTransform},
             },
-            state::{bootstrap_state::ChannelsState, metrics_state::MetricsState},
+            state::{bootstrap_state::ChannelsState, otel_state::OpMetricState},
         };
 
         use super::EndpointBrokerState;
@@ -1617,7 +1621,7 @@ mod tests {
             let (tx, _) = channel(2);
             let client = RippleClient::new(ChannelsState::new());
             let state = EndpointBrokerState::new(
-                MetricsState::default(),
+                OpMetricState::default(),
                 tx,
                 RuleEngine {
                     rules: RuleSet::default(),

core/main/src/broker/rules_engine.rs

@@ -249,10 +249,8 @@ impl RuleEngine {
         }
     }
 
-    pub fn has_rule(&self, request: &RpcRequest) -> bool {
-        self.rules
-            .rules
-            .contains_key(&request.ctx.method.to_lowercase())
+    pub fn has_rule(&self, request: &str) -> bool {
+        self.rules.rules.contains_key(&request.to_lowercase())
     }
 
     pub fn get_rule(&self, rpc_request: &RpcRequest) -> Option<Rule> {

core/main/src/firebolt/firebolt_gateway.rs

@@ -303,11 +303,6 @@ impl FireboltGateway {
             .metrics
             .add_api_stats(&request_c.ctx.request_id, &request_c.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(),
-        );
         let fail_open = matches!(
             platform_state
                 .get_device_manifest()
@@ -322,13 +317,6 @@ impl FireboltGateway {
             capture_stage(&platform_state.metrics, &request_c, "context_ready");
             // Validate incoming request parameters.
             if let Err(error_string) = validate_request(open_rpc_state, &request_c, fail_open) {
-                TelemetryBuilder::stop_and_send_firebolt_metrics_timer(
-                    &platform_state.clone(),
-                    metrics_timer,
-                    format!("{}", JSON_RPC_STANDARD_ERROR_INVALID_PARAMS),
-                )
-                .await;
-
                 let json_rpc_error = JsonRpcError {
                     code: JSON_RPC_STANDARD_ERROR_INVALID_PARAMS,
                     message: error_string,
@@ -418,13 +406,8 @@ impl FireboltGateway {
                                     .emit_debug();
 
                                     // if the websocket disconnects before the session is recieved this leads to an error
-                                    RpcRouter::route(
-                                        platform_state.clone(),
-                                        request_c,
-                                        session,
-                                        metrics_timer.clone(),
-                                    )
-                                    .await;
+                                    RpcRouter::route(platform_state.clone(), request_c, session)
+                                        .await;
                                 } else {
                                     error!("session is missing request is not forwarded for request {:?}", request_c.ctx);
                                 }
@@ -435,12 +418,6 @@ impl FireboltGateway {
                 Err(e) => {
                     let deny_reason = e.reason;
                     // log firebolt response message in RDKTelemetry 1.0 friendly format
-                    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 {:?} {:?}",