main.rs

// TODO: Look into how to properly resolve `clippy::result_large_err`.
// This will need changes in our and upstream error types.
#![allow(clippy::result_large_err)]
use std::{future::Future, sync::Arc};

use anyhow::anyhow;
use clap::Parser;
use futures::{FutureExt, StreamExt, TryFutureExt};
use product_config::ProductConfigManager;
use stackable_operator::{
    YamlSchema,
    cli::{Command, RunArguments},
    client::{self, Client},
    eos::EndOfSupportChecker,
    k8s_openapi::api::{
        apps::v1::DaemonSet,
        core::v1::{ConfigMap, Service},
    },
    kube::{
        Api, CustomResourceExt as _,
        core::DeserializeGuard,
        runtime::{
            Controller,
            events::{Recorder, Reporter},
            watcher,
        },
    },
    logging::controller::report_controller_reconciled,
    namespace::WatchNamespace,
    shared::yaml::SerializeOptions,
    telemetry::Tracing,
    utils::{
        cluster_info::KubernetesClusterInfo,
        signal::{self, SignalWatcher},
    },
};

use crate::{
    controller::OPA_FULL_CONTROLLER_NAME,
    crd::{OPERATOR_NAME, OpaCluster, OpaClusterVersion, v1alpha2},
    webhooks::conversion::create_webhook_server,
};

mod controller;
mod crd;
mod discovery;
mod operations;
mod product_logging;
mod service;
mod webhooks;

pub mod built_info {
    include!(concat!(env!("OUT_DIR"), "/built.rs"));
}

#[derive(Parser)]
#[clap(about, author)]
struct Opts {
    #[clap(subcommand)]
    cmd: Command<OpaRun>,
}

#[derive(clap::Parser)]
struct OpaRun {
    /// The full image tag of the operator, used to deploy the user_info_fetcher.
    #[clap(long, env)]
    operator_image: String,

    #[clap(flatten)]
    common: RunArguments,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let opts = Opts::parse();
    match opts.cmd {
        Command::Crd => {
            OpaCluster::merged_crd(OpaClusterVersion::V1Alpha2)?
                .print_yaml_schema(built_info::PKG_VERSION, SerializeOptions::default())?;
        }
        Command::Run(OpaRun {
            operator_image,
            common:
                RunArguments {
                    operator_environment,
                    watch_namespace,
                    product_config,
                    maintenance,
                    common,
                },
        }) => {
            // As stackable-operator pulls in ring and reqwest >= 0.13 pulls in aws_lc_rs, we need
            // to explicitly tell rustls what provider to use. As other operators use ring, we use
            // that for consistency reasons here as well.
            rustls::crypto::ring::default_provider()
                .install_default()
                .expect("failed to install ring rustls provider");

            // NOTE (@NickLarsenNZ): Before stackable-telemetry was used:
            // - The console log level was set by `OPA_OPERATOR_LOG`, and is now `CONSOLE_LOG` (when using Tracing::pre_configured).
            // - The file log level was set by `OPA_OPERATOR_LOG`, and is now set via `FILE_LOG` (when using Tracing::pre_configured).
            // - The file log directory was set by `OPA_OPERATOR_LOG_DIRECTORY`, and is now set by `ROLLING_LOGS_DIR` (or via `--rolling-logs <DIRECTORY>`).
            let _tracing_guard =
                Tracing::pre_configured(built_info::PKG_NAME, common.telemetry).init()?;

            tracing::info!(
                built_info.pkg_version = built_info::PKG_VERSION,
                built_info.git_version = built_info::GIT_VERSION,
                built_info.target = built_info::TARGET,
                built_info.built_time_utc = built_info::BUILT_TIME_UTC,
                built_info.rustc_version = built_info::RUSTC_VERSION,
                "Starting {description}",
                description = built_info::PKG_DESCRIPTION
            );

            // Watches for the SIGTERM signal and sends a signal to all receivers, which gracefully
            // shuts down all concurrent tasks below (EoS checker, controller).
            let sigterm_watcher = SignalWatcher::sigterm()?;

            let eos_checker =
                EndOfSupportChecker::new(built_info::BUILT_TIME_UTC, maintenance.end_of_support)?
                    .run(sigterm_watcher.handle())
                    .map(anyhow::Ok);

            let product_config = product_config.load(&[
                "deploy/config-spec/properties.yaml",
                "/etc/stackable/opa-operator/config-spec/properties.yaml",
            ])?;

            let client =
                client::initialize_operator(Some(OPERATOR_NAME.to_string()), &common.cluster_info)
                    .await?;

            let kubernetes_cluster_info = client.kubernetes_cluster_info.clone();

            let webhook_server = create_webhook_server(
                &operator_environment,
                maintenance.disable_crd_maintenance,
                client.as_kube_client(),
            )
            .await?;

            let webhook_server = webhook_server
                .run(sigterm_watcher.handle())
                .map_err(|err| anyhow!(err).context("failed to run webhook server"));

            let controller = create_controller(
                client.clone(),
                product_config,
                watch_namespace,
                operator_image.clone(),
                operator_image,
                kubernetes_cluster_info,
                sigterm_watcher.handle(),
            )
            .map(anyhow::Ok);

            let delayed_controller = async {
                signal::crd_established(&client, v1alpha2::OpaCluster::crd_name(), None).await?;
                controller.await
            };

            futures::try_join!(delayed_controller, webhook_server, eos_checker)?;
        }
    };

    Ok(())
}

/// This creates an instance of a [`Controller`] which waits for incoming events and reconciles them.
///
/// This is an async method and the returned future needs to be consumed to make progress.
async fn create_controller<F>(
    client: Client,
    product_config: ProductConfigManager,
    watch_namespace: WatchNamespace,
    opa_bundle_builder_image: String,
    user_info_fetcher_image: String,
    cluster_info: KubernetesClusterInfo,
    shutdown_signal: F,
) where
    F: Future<Output = ()> + Send + Sync + 'static,
{
    let opa_api: Api<DeserializeGuard<v1alpha2::OpaCluster>> = watch_namespace.get_api(&client);
    let daemonsets_api: Api<DeserializeGuard<DaemonSet>> = watch_namespace.get_api(&client);
    let configmaps_api: Api<DeserializeGuard<ConfigMap>> = watch_namespace.get_api(&client);
    let services_api: Api<DeserializeGuard<Service>> = watch_namespace.get_api(&client);

    let controller = Controller::new(opa_api, watcher::Config::default())
        .owns(daemonsets_api, watcher::Config::default())
        .owns(configmaps_api, watcher::Config::default())
        .owns(services_api, watcher::Config::default());

    let event_recorder = Arc::new(Recorder::new(
        client.as_kube_client(),
        Reporter {
            controller: OPA_FULL_CONTROLLER_NAME.to_string(),
            instance: None,
        },
    ));
    controller
        .graceful_shutdown_on(shutdown_signal)
        .run(
            controller::reconcile_opa,
            controller::error_policy,
            Arc::new(controller::Ctx {
                client: client.clone(),
                product_config,
                opa_bundle_builder_image,
                user_info_fetcher_image,
                cluster_info,
            }),
        )
        // We can let the reporting happen in the background
        .for_each_concurrent(
            16, // concurrency limit
            |result| {
                // The event_recorder needs to be shared across all invocations, so that
                // events are correctly aggregated
                let event_recorder = event_recorder.clone();
                async move {
                    report_controller_reconciled(
                        &event_recorder,
                        OPA_FULL_CONTROLLER_NAME,
                        &result,
                    )
                    .await;
                }
            },
        )
        .await;
}