You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: k8.rs
    path of file: ./repos/fluvio/crates/fluvio-cluster/src/start
    
    the code of the file until where you have to start completion: use std::io::BufReader;
use std::io::Error as IoError;
use std::io::ErrorKind;
use std::collections::BTreeMap;
use std::io::Read;
use std::path::PathBuf;
use std::borrow::Cow;
use std::process::Child;
use std::process::Command;
use std::time::Duration;
use std::env;
use std::time::SystemTime;

use anyhow::{anyhow, Result};
use derive_builder::Builder;
use k8_client::meta_client::NameSpace;
use tracing::{info, warn, debug, instrument};
use once_cell::sync::Lazy;
use tempfile::NamedTempFile;
use semver::Version;

use fluvio::FluvioAdmin;
use fluvio_controlplane_metadata::spg::SpuConfig;
use fluvio_sc_schema::objects::CommonCreateRequest;
use fluvio_types::defaults::TLS_CLIENT_SECRET_NAME;
use fluvio_types::defaults::TLS_SERVER_SECRET_NAME;
use k8_client::SharedK8Client;
use k8_client::load_and_share;
use k8_types::K8Obj;
use k8_types::app::deployment::DeploymentSpec;
use fluvio::{Fluvio, FluvioConfig};
use fluvio::metadata::spg::SpuGroupSpec;
use fluvio::metadata::spu::SpuSpec;
use fluvio::config::{TlsPolicy, TlsConfig, TlsPaths, ConfigFile};
use fluvio_future::timer::sleep;
use k8_config::K8Config;
use k8_client::meta_client::MetadataClient;
use k8_types::core::service::{LoadBalancerType, ServiceSpec, TargetPort};
use k8_types::core::node::{NodeSpec, NodeAddress};
use fluvio_command::CommandExt;

use crate::InstallationType;
use crate::check::{AlreadyInstalled, SysChartCheck};
use crate::error::K8InstallError;
use crate::progress::ProgressBarFactory;
use crate::render::ProgressRenderedText;
use crate::render::ProgressRenderer;
use crate::start::common::check_crd;
use crate::tls_config_to_cert_paths;
use crate::{StartStatus, DEFAULT_NAMESPACE, ClusterChecker};
use crate::charts::{ChartConfig, ChartInstaller};
use crate::UserChartLocation;
use crate::progress::InstallProgressMessage;

use super::constants::*;
use super::common::try_connect_to_sc;

const DEFAULT_REGISTRY: &str = "infinyon";
const DEFAULT_GROUP_NA
    
    pub async fn install_fluvio(&self) -> Result<StartStatus> {
        if !self.config.skip_checks {
            self.preflight_check(true).await?;
        }

        self.install_app().await?;

        // before we do let's try make sure SPU are installed.
        check_crd(self.kube_client.clone()).await?;

        let pb = self.pb_factory.create()?;

        let sc_service = self.discover_sc_service().await?;
        let (external_host, external_port) =
            self.discover_sc_external_host_and_port(&sc_service).await?;
        let external_host_and_port = format!("{external_host}:{external_port}");

        if self.config.save_profile {
            self.update_profile(&external_host_and_port)?;
        }

        self.wait_for_sc_availability().await?;

        let (install_host_and_port, pf_process) = if self.config.use_k8_port_forwarding {
            pb.println("Using K8 port forwarding for install".to_string());
            let (install_host, install_port, pf_process) =
                self.start_sc_port_forwarding(&sc_service, &pb).await?;
            (format!("{install_host}:{install_port}"), Some(pf_process))
        } else {
            (external_host_and_port.clone(), None)
        };

        let cluster_config = FluvioConfig::new(install_host_and_port.clone())
            .with_tls(self.config.client_tls_policy.clone());
        pb.set_message("üîé Discovering Fluvio SC");
        let fluvio =
            match try_connect_to_sc(&cluster_config, &self.config.platform_version, &pb).await {
                Some(fluvio) => fluvio,
                None => return Err(K8InstallError::SCServiceTimeout.into()),
            };
        pb.println(format!("‚úÖ Connected to SC: {install_host_and_port}"));
        pb.finish_and_clear();
        drop(pb);

        // Create a managed SPU cluster
        self.create_managed_spu_group(&fluvio).await?;

        if let Some(mut pf_process) = pf_process {
            match pf_process.kill() {
                Ok(_) => info!("Port forwarding process exited normally"),
                Err(_) => warn!("Port forwarding process terminated prematurely"),
            };
        }

        self.pb_factory
            .println(InstallProgressMessage::Success.msg());

        Ok(StartStatus {
            address: external_host,
            port: external_port,
        })
    }