streaming_telemetry/src/main.rs

// Copyright 2025 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

mod app_state;
mod composite_query;
mod dbus_client;
mod grpc;
mod handlers;
#[cfg(feature = "mtls")]
mod mtls;
mod telemetry_source_manager;
use crate::grpc::telemetry_server::{load_server_config, BmcTelemetryService};
use crate::grpc::third_party_voyager::machine_telemetry_server::MachineTelemetryServer;

use crate::app_state::AppState;
#[cfg(feature = "mtls")]
use crate::mtls::mtls::run_secure_server;
use clap::{arg, Command};
use handlers::chassis::ChassisState;
use std::sync::Arc;
use telemetry_source_manager::telemetry_source_manager_api::create_telemetry_source_manager;

fn cli() -> Command {
    Command::new("my_server")
        .about("Starts a server with the specified configuration")
        .arg(
            arg!(-c --config <CONFIG> "Streaming telemetry server config textproto file")
                .default_value("server_config.textproto")
                .value_parser(clap::value_parser!(String)),
        )
        .arg(
            arg!(-i --insecure "Disables SSL certificate verification")
                .required(false)
                .default_value("false")
                .action(clap::ArgAction::SetTrue),
        )
        .arg(
            arg!(-k --key <KEY> "Server's private key file name")
                .required_if_eq("insecure", "false")
                .value_parser(clap::value_parser!(String)),
        )
        .arg(
            arg!(-c --cert <CERT> "Server's certificate file name")
                .required_if_eq("insecure", "false")
                .value_parser(clap::value_parser!(String)),
        )
        .arg(
            arg!(-a --cacert <CACERT> "CA certificate chain file name")
                .required_if_eq("insecure", "false")
                .value_parser(clap::value_parser!(String)),
        )
        .arg(
            arg!(-p --policy <POLICY> "Client authentication policy file name")
                .required_if_eq("insecure", "false")
                .value_parser(clap::value_parser!(String)),
        )
        .arg(
            arg!(-s --crls <CRLS> "Certificate Revocation List file name")
                .required(false)
                .value_parser(clap::value_parser!(String)),
        )
        .arg(
            arg!(-P --port <PORT> "Server's listening port number")
                .required(true)
                .value_parser(clap::value_parser!(u16)),
        )
        .arg(
            arg!(-e --emconfig <EMCONFIG> "EntityManager cofnig files separated by comma")
                .required(true)
                .value_parser(clap::value_parser!(String)),
        )
}

async fn run_insecure_server(
    port: u16,
    grpc: MachineTelemetryServer<BmcTelemetryService>,
) -> Result<(), Box<dyn std::error::Error>> {
    let addr = format!("[::]:{port}").parse()?;
    tonic::transport::Server::builder()
        .add_service(grpc)
        .serve(addr)
        .await?;

    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    env_logger::init();

    let matches = cli().get_matches();

    let port = matches
        .get_one::<u16>("port")
        .expect("required argument")
        .to_owned();

    let server_config = matches.get_one::<String>("config").unwrap();
    let server_config = load_server_config(server_config).unwrap();
    // The server_config will not be read at data path, means when data streaming started, the
    // application will not try to access it anymore
    let server_config = Arc::new(tokio::sync::RwLock::new(server_config));

    let em_config = matches.get_one::<String>("emconfig").unwrap();
    let em_config: Vec<String> = em_config.split(',').map(|s| s.trim().to_string()).collect();

    let chassis_state = ChassisState::default();
    let telemetry_source_manager =
        create_telemetry_source_manager(1_000, 10_000, server_config.clone(), em_config).await?;
    let state = Arc::new(AppState {
        chassis_state,
        telemetry_source_manager,
    });

    let grpc = MachineTelemetryServer::new(BmcTelemetryService {
        state,
        server_config,
    });

    if matches.get_flag("insecure") {
        println!("Running in insecure mode on port {}...", port);
        run_insecure_server(port, grpc).await
    } else {
        #[cfg(feature = "mtls")]
        {
            let key = matches.get_one::<String>("key").unwrap();
            let cert = matches.get_one::<String>("cert").unwrap();
            let cacert = matches.get_one::<String>("cacert").unwrap();
            let policy = matches.get_one::<String>("policy").unwrap();
            let crls = matches.get_one::<String>("crls").map(|s| s.as_str());

            println!("Starting secure server on port {}...", port);
            println!("Using key file: {}", key);
            println!("Using certificate file: {}", cert);
            println!("Using CA certificate file: {}", cacert);
            println!("Using policy file: {}", policy);
            run_secure_server(port, key, cert, cacert, policy, crls, grpc).await?;
        }
        Ok(())
    }
}