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::sync::Arc;

use clap::Parser;
use crd::v1alpha1::NifiClusteringBackend;
use futures::{FutureExt, StreamExt};
use stackable_operator::{
    YamlSchema,
    cli::{Command, RunArguments},
    crd::authentication::core as auth_core,
    eos::EndOfSupportChecker,
    k8s_openapi::api::{
        apps::v1::StatefulSet,
        core::v1::{ConfigMap, Service},
    },
    kube::{
        ResourceExt,
        core::DeserializeGuard,
        runtime::{
            Controller,
            events::{Recorder, Reporter},
            reflector::ObjectRef,
            watcher,
        },
    },
    logging::controller::report_controller_reconciled,
    shared::yaml::SerializeOptions,
    telemetry::Tracing,
    utils::signal::SignalWatcher,
};

use crate::{
    controller::NIFI_FULL_CONTROLLER_NAME,
    crd::{NifiCluster, NifiClusterVersion, authorization::NifiOpaConfig, v1alpha1},
};

mod config;
mod controller;
mod crd;
mod listener;
mod operations;
mod product_logging;
mod reporting_task;
mod security;
mod service;

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

const OPERATOR_NAME: &str = "nifi.stackable.tech";

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

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let opts = Opts::parse();
    match opts.cmd {
        Command::Crd => NifiCluster::merged_crd(NifiClusterVersion::V1Alpha1)?
            .print_yaml_schema(built_info::PKG_VERSION, SerializeOptions::default())?,
        Command::Run(RunArguments {
            operator_environment: _,
            watch_namespace,
            product_config,
            maintenance,
            common,
        }) => {
            // NOTE (@NickLarsenNZ): Before stackable-telemetry was used:
            // - The console log level was set by `NIFI_OPERATOR_LOG`, and is now `CONSOLE_LOG` (when using Tracing::pre_configured).
            // - The file log level was set by `NIFI_OPERATOR_LOG`, and is now set via `FILE_LOG` (when using Tracing::pre_configured).
            // - The file log directory was set by `NIFI_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/nifi-operator/config-spec/properties.yaml",
            ])?;

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

            let event_recorder = Arc::new(Recorder::new(
                client.as_kube_client(),
                Reporter {
                    controller: NIFI_FULL_CONTROLLER_NAME.to_string(),
                    instance: None,
                },
            ));

            let nifi_controller = Controller::new(
                watch_namespace.get_api::<DeserializeGuard<v1alpha1::NifiCluster>>(&client),
                watcher::Config::default(),
            );

            let authentication_class_store = nifi_controller.store();
            let config_map_store = nifi_controller.store();

            let nifi_controller = nifi_controller
                .owns(
                    watch_namespace.get_api::<Service>(&client),
                    watcher::Config::default(),
                )
                .owns(
                    watch_namespace.get_api::<StatefulSet>(&client),
                    watcher::Config::default(),
                )
                .owns(
                    watch_namespace.get_api::<ConfigMap>(&client),
                    watcher::Config::default(),
                )
                .watches(
                    client
                        .get_api::<DeserializeGuard<auth_core::v1alpha1::AuthenticationClass>>(&()),
                    watcher::Config::default(),
                    move |_| {
                        authentication_class_store
                            .state()
                            .into_iter()
                            .map(|nifi| ObjectRef::from_obj(&*nifi))
                    },
                )
                .watches(
                    watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
                    watcher::Config::default(),
                    move |config_map| {
                        config_map_store
                            .state()
                            .into_iter()
                            .filter(move |nifi| references_config_map(nifi, &config_map))
                            .map(|nifi| ObjectRef::from_obj(&*nifi))
                    },
                )
                .graceful_shutdown_on(sigterm_watcher.handle())
                .run(
                    controller::reconcile_nifi,
                    controller::error_policy,
                    Arc::new(controller::Ctx {
                        client: client.clone(),
                        product_config,
                    }),
                )
                // We can let the reporting happen in the background
                .for_each_concurrent(
                    16, // concurrency limit
                    move |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,
                                NIFI_FULL_CONTROLLER_NAME,
                                &result,
                            )
                            .await;
                        }
                    },
                )
                .map(anyhow::Ok);

            futures::try_join!(nifi_controller, eos_checker)?;
        }
    }

    Ok(())
}

fn references_config_map(
    nifi: &DeserializeGuard<v1alpha1::NifiCluster>,
    config_map: &DeserializeGuard<ConfigMap>,
) -> bool {
    let Ok(nifi) = &nifi.0 else {
        return false;
    };

    let references_zookeeper_config_map = match &nifi.spec.cluster_config.clustering_backend {
        NifiClusteringBackend::ZooKeeper {
            zookeeper_config_map_name,
        } => *zookeeper_config_map_name == config_map.name_any(),
        NifiClusteringBackend::Kubernetes {} => false,
    };
    let references_authorization_config_map = references_authorization_config_map(nifi, config_map);

    references_zookeeper_config_map || references_authorization_config_map
}

fn references_authorization_config_map(
    nifi: &v1alpha1::NifiCluster,
    config_map: &DeserializeGuard<ConfigMap>,
) -> bool {
    match &nifi.spec.cluster_config.authorization {
        crd::authorization::NifiAuthorization::Opa {
            opa: NifiOpaConfig { opa, .. },
        } => opa.config_map_name == config_map.name_any(),
        crd::authorization::NifiAuthorization::SingleUser {} => false,
        crd::authorization::NifiAuthorization::Standard { .. } => false,
    }
}