CruiseControl extends AbstractModel { public static CruiseControl fromCrd(Kafka kafkaAssembly, KafkaVersion.Lookup versions) { CruiseControl cruiseControl = null; CruiseControlSpec spec = kafkaAssembly.getSpec().getCruiseControl(); KafkaClusterSpec kafkaClusterSpec = kafkaAssembly.getSpec().getKafka(); if (spec != null) { cruiseControl = new CruiseControl(kafkaAssembly); cruiseControl.isDeployed = true; cruiseControl.setReplicas(DEFAULT_REPLICAS); String image = spec.getImage(); if (image == null) { image = System.getenv().getOrDefault(ClusterOperatorConfig.STRIMZI_DEFAULT_CRUISE_CONTROL_IMAGE, versions.kafkaImage(kafkaClusterSpec.getImage(), versions.defaultVersion().version())); } cruiseControl.setImage(image); TlsSidecar tlsSidecar = spec.getTlsSidecar(); if (tlsSidecar == null) { tlsSidecar = new TlsSidecar(); } String tlsSideCarImage = tlsSidecar.getImage(); if (tlsSideCarImage == null) { tlsSideCarImage = System.getenv().getOrDefault(ClusterOperatorConfig.STRIMZI_DEFAULT_TLS_SIDECAR_CRUISE_CONTROL_IMAGE, versions.kafkaImage(kafkaClusterSpec.getImage(), versions.defaultVersion().version())); } tlsSidecar.setImage(tlsSideCarImage); cruiseControl.tlsSidecarImage = tlsSideCarImage; cruiseControl.setTlsSidecar(tlsSidecar); cruiseControl = updateConfiguration(spec, cruiseControl); KafkaConfiguration configuration = new KafkaConfiguration(kafkaClusterSpec.getConfig().entrySet()); if (configuration.getConfigOption(MIN_INSYNC_REPLICAS) != null) { cruiseControl.minInsyncReplicas = configuration.getConfigOption(MIN_INSYNC_REPLICAS); } Capacity capacity = new Capacity(kafkaAssembly.getSpec()); cruiseControl.brokerDiskMiBCapacity = capacity.getDiskMiB(); cruiseControl.brokerCpuUtilizationCapacity = capacity.getCpuUtilization(); cruiseControl.brokerInboundNetworkKiBPerSecondCapacity = capacity.getInboundNetworkKiBPerSecond(); cruiseControl.brokerOuboundNetworkKiBPerSecondCapacity = capacity.getOutboundNetworkKiBPerSecond(); Map<String, Object> metrics = spec.getMetrics(); if (metrics != null) { cruiseControl.setMetricsEnabled(true); cruiseControl.setMetricsConfig(metrics.entrySet()); } if (spec.getReadinessProbe() != null) { cruiseControl.setReadinessProbe(spec.getReadinessProbe()); } if (spec.getLivenessProbe() != null) { cruiseControl.setLivenessProbe(spec.getLivenessProbe()); } Logging logging = spec.getLogging(); cruiseControl.setLogging(logging == null ? new InlineLogging() : logging); cruiseControl.setGcLoggingEnabled(spec.getJvmOptions() == null ? DEFAULT_JVM_GC_LOGGING_ENABLED : spec.getJvmOptions().isGcLoggingEnabled()); cruiseControl.setJvmOptions(spec.getJvmOptions()); cruiseControl.setResources(spec.getResources()); cruiseControl.setOwnerReference(kafkaAssembly); cruiseControl = updateTemplate(spec, cruiseControl); } return cruiseControl; } protected  CruiseControl(HasMetadata resource); static String metricAndLogConfigsName(String cluster); static CruiseControl fromCrd(Kafka kafkaAssembly, KafkaVersion.Lookup versions); static CruiseControl updateConfiguration(CruiseControlSpec spec, CruiseControl cruiseControl); static void checkGoals(CruiseControlConfiguration configuration); static CruiseControl updateTemplate(CruiseControlSpec spec, CruiseControl cruiseControl); static String cruiseControlName(String cluster); static String cruiseControlServiceName(String cluster); Service generateService(); Deployment generateDeployment(boolean isOpenShift, Map<String, String> annotations, ImagePullPolicy imagePullPolicy, List<LocalObjectReference> imagePullSecrets); PodDisruptionBudget generatePodDisruptionBudget(); static String cruiseControlServiceAccountName(String cluster); static String secretName(String kafkaCluster); Secret generateSecret(ClusterCa clusterCa, boolean isMaintenanceTimeWindowsSatisfied); static String policyName(String cluster); NetworkPolicy generateNetworkPolicy(boolean namespaceAndPodSelectorNetworkPolicySupported); static final String CRUISE_CONTROL_METRIC_REPORTER; static final String ANNO_STRIMZI_IO_LOGGING; static final String ENV_VAR_CRUISE_CONTROL_METRICS_ENABLED; static final boolean DEFAULT_CRUISE_CONTROL_METRICS_ENABLED; static final Probe DEFAULT_HEALTHCHECK_OPTIONS; static final String REST_API_PORT_NAME; static final int REST_API_PORT; static final String MIN_INSYNC_REPLICAS; }
@Test public void testGoalsCheck() { String customGoals = "com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal," + "com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal"; Map<String, Object> customGoalConfig = (Map) configuration.asOrderedProperties().asMap(); customGoalConfig.put(CRUISE_CONTROL_DEFAULT_GOALS_CONFIG_KEY.getName(), customGoals); CruiseControlSpec ccSpecWithCustomGoals = new CruiseControlSpecBuilder() .withImage(ccImage) .withConfig(customGoalConfig) .build(); Kafka resourceWithCustomGoals = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout)) .editSpec() .editKafka() .withVersion(version) .withConfig(kafkaConfig) .endKafka() .withCruiseControl(ccSpecWithCustomGoals) .endSpec() .build(); CruiseControl cruiseControlWithCustomGoals = CruiseControl.fromCrd(resourceWithCustomGoals, VERSIONS); String anomalyDetectionGoals = cruiseControlWithCustomGoals .getConfiguration().asOrderedProperties().asMap() .get(CRUISE_CONTROL_ANOMALY_DETECTION_CONFIG_KEY.getName()); assertThat(anomalyDetectionGoals, is(customGoals)); }