pub fn generate_k8_stateful(
        spg_spec: &SpuGroupSpec,
        group_name: &str,
        group_svc_name: &str,
        namespace: &str,
        spu_k8_config: &ScK8Config,
        tls_config: Option<&TlsConfig>,
    ) -> K8StatefulSetSpec {
        let replicas = spg_spec.replicas;
        let spu_template = &spg_spec.spu_config;
        let mut public_port = ContainerPortSpec {
            container_port: SPU_PUBLIC_PORT,
            ..Default::default()
        };
        public_port.name = Some("public".to_owned());

        let mut private_port = ContainerPortSpec {
            container_port: SPU_PRIVATE_PORT,
            ..Default::default()
        };
        private_port.name = Some("private".to_owned());

        // storage is special because defaults are explicit.
        let storage = spu_template.real_storage_config();
        let size = storage.size;
        let mut env = vec![
            Env::key_field_ref("SPU_INDEX", "metadata.name"),
            Env::key_value("SPU_MIN", &format!("{}", spg_spec.min_id)),
        ];

        let mut volume_mounts = vec![VolumeMount {
            name: "data".to_owned(),
            mount_path: format!("/var/lib/{}/data", PRODUCT_NAME),
            ..Default::default()
        }];

        let mut volumes = vec![];

        let mut args = vec![
            "/fluvio-run".to_owned(),
            "spu".to_owned(),
            "--sc-addr".to_owned(),
            format!(
                "fluvio-sc-internal.{}.svc.cluster.local:{}",
                namespace, SC_PRIVATE_PORT
            ),
            "--log-base-dir".to_owned(),
            storage.log_dir,
            "--log-size".to_owned(),
            size.clone(),
        ];

        if let Some(tls) = tls_config {
            args.push("--tls".to_owned());
            if tls.enable_client_cert {
                args.push("--enable-client-cert".to_owned());
                args.push("--ca-cert".to_owned());
                args.push(tls.ca_cert.clone().unwrap());
                volume_mounts.push(VolumeMount {
                    name: "cacert".to_owned(),
                    mount_path: "/var/certs/ca".to_owned(),
                    read_only: Some(true),
                    ..Default::default()
                });
                volumes.push(VolumeSpec {
                    name: "cacert".to_owned(),
                    secret: Some(SecretVolumeSpec {
                        secret_name: "fluvio-ca".to_owned(), // fixed
                        ..Default::default()
                    }),
                    ..Default::default()
                });
            }

            args.push("--server-cert".to_owned());
            args.push(tls.server_cert.clone().unwrap());
            args.push("--server-key".to_owned());
            args.push(tls.server_key.clone().unwrap());

            volume_mounts.push(VolumeMount {
                name: "tls".to_owned(),
                mount_path: "/var/certs/tls".to_owned(),
                read_only: Some(true),
                ..Default::default()
            });

            volumes.push(VolumeSpec {
                name: "tls".to_owned(),
                secret: Some(SecretVolumeSpec {
                    secret_name: "fluvio-tls".to_owned(), // fixed
                    ..Default::default()
                }),
                ..Default::default()
            });

            args.push("--bind-non-tls-public".to_owned());
            args.push("0.0.0.0:9007".to_owned());
        }

        // add RUST LOG, if passed
        if let Ok(rust_log) = std::env::var("RUST_LOG") {
            env.push(Env::key_value("RUST_LOG", &rust_log));
        }

        // env.append(&mut spu_template.env.clone());

        let spu_pod_config = &spu_k8_config.spu_pod_config;

        let template = TemplateSpec {
            metadata: Some(
                TemplateMeta::default()
                    .set_labels(vec![("app", SPU_DEFAULT_NAME), ("group", group_name)]),
            ),
            spec: PodSpec {
                termination_grace_period_seconds: Some(10),
                containers: vec![ContainerSpec {
                    name: SPU_DEFAULT_NAME.to_owned(),
                    image: Some(spu_k8_config.image.clone()),
                    resources: spu_pod_config.resources.clone(),
                    ports: vec![public_port, private_port],
                    volume_mounts,
                    env,
                    args,
                    ..Default::default()
                }],
                volumes,
                security_context: spu_k8_config.pod_security_context.clone(),
                node_selector: Some(spu_pod_config.node_selector.clone()),
                ..Default::default()
            },
        };
        let claim = PersistentVolumeClaim {
            access_modes: vec![VolumeAccessMode::ReadWriteOnce],
            storage_class_name: spu_pod_config.storage_class.clone(),
            resources: ResourceRequirements {
                requests: VolumeRequest { storage: size },
            },
        };

        K8StatefulSetSpec {
            replicas: Some(replicas),
            service_name: group_svc_name.to_owned(),
            selector: LabelSelector::new_labels(vec![
                ("app", SPU_DEFAULT_NAME),
                ("group", group_name),
            ]),
            template,
            volume_claim_templates: vec![TemplateSpec {
                spec: claim,
                metadata: Some(TemplateMeta::named("data")),
            }],
            ..Default::default()
        }
    }
/// convert spu group spec into k8 statefulset spec
convert spu group spec into k8 statefulset spec
convert spu group spec into k8 statefulset spec
pub fn generate_k8_stateful(
        spg_spec: &SpuGroupSpec,
        group_name: &str,
        group_svc_name: &str,
        namespace: &str,
        spu_k8_config: &ScK8Config,
        tls_config: Option<&TlsConfig>,
    ) -> K8StatefulSetSpec {
        let replicas = spg_spec.replicas;
        let spu_template = &spg_spec.spu_config;
        let mut public_port = ContainerPortSpec {
            container_port: SPU_PUBLIC_PORT,
            ..Default::default()
        };
        public_port.name = Some("public".to_owned());
        let mut private_port = ContainerPortSpec {
            container_port: SPU_PRIVATE_PORT,
            ..Default::default()
        };
        private_port.name = Some("private".to_owned());
        // storage is special because defaults are explicit.
        let storage = spu_template.real_storage_config();
        let size = storage.size;
        let mut env = vec![
            Env::key_field_ref("SPU_INDEX", "metadata.name"),
            Env::key_value("SPU_MIN", &format!("{}", spg_spec.min_id)),
        ];
        let mut volume_mounts = vec![VolumeMount {
            name: "data".to_owned(),
            mount_path: format!("/var/lib/{}/data", PRODUCT_NAME),
            ..Default::default()
        }];
        let mut volumes = vec![];
        let mut args = vec![
            "/fluvio-run".to_owned(),
            "spu".to_owned(),
            "--sc-addr".to_owned(),
            format!(
                "fluvio-sc-internal.{}.svc.cluster.local:{}",
                namespace, SC_PRIVATE_PORT
            ),
            "--log-base-dir".to_owned(),
            storage.log_dir,
            "--log-size".to_owned(),
            size.clone(),
        ];
        if let Some(tls) = tls_config {
            args.push("--tls".to_owned());
            if tls.enable_client_cert {
                args.push("--enable-client-cert".to_owned());
                args.push("--ca-cert".to_owned());
                args.push(tls.ca_cert.clone().unwrap());
                volume_mounts.push(VolumeMount {
                    name: "cacert".to_owned(),
                    mount_path: "/var/certs/ca".to_owned(),
                    read_only: Some(true),
                    ..Default::default()
                });
                volumes.push(VolumeSpec {
                    name: "cacert".to_owned(),
                    secret: Some(SecretVolumeSpec {
                        secret_name: "fluvio-ca".to_owned(), // fixed
                        ..Default::default()
                    }),
                    ..Default::default()
                });
            }
            args.push("--server-cert".to_owned());
            args.push(tls.server_cert.clone().unwrap());
            args.push("--server-key".to_owned());
            args.push(tls.server_key.clone().unwrap());
            volume_mounts.push(VolumeMount {
                name: "tls".to_owned(),
                mount_path: "/var/certs/tls".to_owned(),
                read_only: Some(true),
                ..Default::default()
            });
            volumes.push(VolumeSpec {
                name: "tls".to_owned(),
                secret: Some(SecretVolumeSpec {
                    secret_name: "fluvio-tls".to_owned(), // fixed
                    ..Default::default()
                }),
                ..Default::default()
            });
            args.push("--bind-non-tls-public".to_owned());
            args.push("0.0.0.0:9007".to_owned());
        }
        // add RUST LOG, if passed
        if let Ok(rust_log) = std::env::var("RUST_LOG") {
            env.push(Env::key_value("RUST_LOG", &rust_log));
        }
        // env.append(&mut spu_template.env.clone());
        let spu_pod_config = &spu_k8_config.spu_pod_config;
        let template = TemplateSpec {
            metadata: Some(
                TemplateMeta::default()
                    .set_labels(vec![("app", SPU_DEFAULT_NAME), ("group", group_name)]),
            ),
            spec: PodSpec {
                termination_grace_period_seconds: Some(10),
                containers: vec![ContainerSpec {
                    name: SPU_DEFAULT_NAME.to_owned(),
                    image: Some(spu_k8_config.image.clone()),
                    resources: spu_pod_config.resources.clone(),
                    ports: vec![public_port, private_port],
                    volume_mounts,
                    env,
                    args,
                    ..Default::default()
                }],
                volumes,
                security_context: spu_k8_config.pod_security_context.clone(),
                node_selector: Some(spu_pod_config.node_selector.clone()),
                ..Default::default()
            },
        };
        let claim = PersistentVolumeClaim {
            access_modes: vec![VolumeAccessMode::ReadWriteOnce],
            storage_class_name: spu_pod_config.storage_class.clone(),
            resources: ResourceRequirements {
                requests: VolumeRequest { storage: size },
            },
        };
        K8StatefulSetSpec {
            replicas: Some(replicas),
            service_name: group_svc_name.to_owned(),
            selector: LabelSelector::new_labels(vec![
                ("app", SPU_DEFAULT_NAME),
                ("group", group_name),
            ]),
            template,
            volume_claim_templates: vec![TemplateSpec {
                spec: claim,
                metadata: Some(TemplateMeta::named("data")),
            }],
            ..Default::default()
        }
    }
pub fn generate_k8_stateful ( spg_spec : & SpuGroupSpec , group_name : & str , group_svc_name : & str , namespace : & str , spu_k8_config : & ScK8Config , tls_config : Option < & TlsConfig > , ) -> K8StatefulSetSpec { let replicas = spg_spec . replicas ; let spu_template = & spg_spec . spu_config ; let mut public_port = ContainerPortSpec { container_port : SPU_PUBLIC_PORT , .. Default :: default ( ) } ; public_port . name = Some ( " " . to_owned ( ) ) ; let mut private_port = ContainerPortSpec { container_port : SPU_PRIVATE_PORT , .. Default :: default ( ) } ; private_port . name = Some ( " " . to_owned ( ) ) ; // storage is special because defaults are explicit. let storage = spu_template . real_storage_config ( ) ; let size = storage . size ; let mut env = vec ! [ Env key_field_ref ( " " " " ) Env key_value ( " " format ( " " spg_spec min_id ) ) ] ; let mut volume_mounts = vec ! [ VolumeMount { name " " to_owned ( ) mount_path format ( " " PRODUCT_NAME ) Default default ( ) } ] ; let mut volumes = vec ! [ ] ; let mut args = vec ! [ " " to_owned ( ) " " to_owned ( ) " " to_owned ( ) format ( " " namespace SC_PRIVATE_PORT ) " " to_owned ( ) storage log_dir " " to_owned ( ) size clone ( ) ] ; if let Some ( tls ) = tls_config { args . push ( " " . to_owned ( ) ) ; if tls . enable_client_cert { args . push ( " " . to_owned ( ) ) ; args . push ( " " . to_owned ( ) ) ; args . push ( tls . ca_cert . clone ( ) . unwrap ( ) ) ; volume_mounts . push ( VolumeMount { name : " " . to_owned ( ) , mount_path : " " . to_owned ( ) , read_only : Some ( true ) , .. Default :: default ( ) } ) ; volumes . push ( VolumeSpec { name : " " . to_owned ( ) , secret : Some ( SecretVolumeSpec { secret_name : " " . to_owned ( ) , // fixed .. Default :: default ( ) } ) , .. Default :: default ( ) } ) ; } args . push ( " " . to_owned ( ) ) ; args . push ( tls . server_cert . clone ( ) . unwrap ( ) ) ; args . push ( " " . to_owned ( ) ) ; args . push ( tls . server_key . clone ( ) . unwrap ( ) ) ; volume_mounts . push ( VolumeMount { name : " " . to_owned ( ) , mount_path : " " . to_owned ( ) , read_only : Some ( true ) , .. Default :: default ( ) } ) ; volumes . push ( VolumeSpec { name : " " . to_owned ( ) , secret : Some ( SecretVolumeSpec { secret_name : " " . to_owned ( ) , // fixed .. Default :: default ( ) } ) , .. Default :: default ( ) } ) ; args . push ( " " . to_owned ( ) ) ; args . push ( " " . to_owned ( ) ) ; } // add RUST LOG, if passed if let Ok ( rust_log ) = std :: env :: var ( " " ) { env . push ( Env :: key_value ( " " , & rust_log ) ) ; } // env.append(&mut spu_template.env.clone()); let spu_pod_config = & spu_k8_config . spu_pod_config ; let template = TemplateSpec { metadata : Some ( TemplateMeta :: default ( ) . set_labels ( vec ! [ ( " " SPU_DEFAULT_NAME ) ( " " group_name ) ] ) , ) , spec : PodSpec { termination_grace_period_seconds : Some ( 10 ) , containers : vec ! [ ContainerSpec { name SPU_DEFAULT_NAME to_owned ( ) image Some ( spu_k8_config image clone ( ) ) resources spu_pod_config resources clone ( ) ports vec [ public_port private_port ] volume_mounts env args Default default ( ) } ] , volumes , security_context : spu_k8_config . pod_security_context . clone ( ) , node_selector : Some ( spu_pod_config . node_selector . clone ( ) ) , .. Default :: default ( ) } , } ; let claim = PersistentVolumeClaim { access_modes : vec ! [ VolumeAccessMode ReadWriteOnce ] , storage_class_name : spu_pod_config . storage_class . clone ( ) , resources : ResourceRequirements { requests : VolumeRequest { storage : size } , } , } ; K8StatefulSetSpec { replicas : Some ( replicas ) , service_name : group_svc_name . to_owned ( ) , selector : LabelSelector :: new_labels ( vec ! [ ( " " SPU_DEFAULT_NAME ) ( " " group_name ) ] ) , template , volume_claim_templates : vec ! [ TemplateSpec { spec claim metadata Some ( TemplateMeta named ( " " ) ) } ] , .. Default :: default ( ) } }
convert spu group spec into k8 statefulset spec
convert spu group spec into k8 statefulset spec
spg_spec
&SpuGroupSpec
group_name
&str
group_svc_name
&str
namespace
&str
spu_k8_config
&ScK8Config
tls_config
Option<&TlsConfig>
spg_spec
&SpuGroupSpec
group_name
&str
group_svc_name
&str
namespace
&str
spu_k8_config
&ScK8Config
tls_config
Option<&TlsConfig>