CruiseControl extends AbstractModel { public NetworkPolicy generateNetworkPolicy(boolean namespaceAndPodSelectorNetworkPolicySupported) { List<NetworkPolicyIngressRule> rules = new ArrayList<>(1); NetworkPolicyIngressRule restApiRule = new NetworkPolicyIngressRuleBuilder() .addNewPort() .withNewPort(REST_API_PORT) .endPort() .build(); if (namespaceAndPodSelectorNetworkPolicySupported) { NetworkPolicyPeer clusterOperatorPeer = new NetworkPolicyPeerBuilder() .withNewPodSelector() .addToMatchLabels(Labels.STRIMZI_KIND_LABEL, "cluster-operator") .endPodSelector() .withNewNamespaceSelector() .endNamespaceSelector() .build(); restApiRule.setFrom(Collections.singletonList(clusterOperatorPeer)); } rules.add(restApiRule); if (isMetricsEnabled) { NetworkPolicyPort metricsPort = new NetworkPolicyPort(); metricsPort.setPort(new IntOrString(METRICS_PORT)); NetworkPolicyIngressRule metricsRule = new NetworkPolicyIngressRuleBuilder() .withPorts(metricsPort) .withFrom() .build(); rules.add(metricsRule); } NetworkPolicy networkPolicy = new NetworkPolicyBuilder() .withNewMetadata() .withName(policyName(cluster)) .withNamespace(namespace) .withLabels(labels.toMap()) .withOwnerReferences(createOwnerReference()) .endMetadata() .withNewSpec() .withNewPodSelector() .addToMatchLabels(Labels.STRIMZI_NAME_LABEL, cruiseControlName(cluster)) .endPodSelector() .withIngress(rules) .endSpec() .build(); log.trace("Created network policy {}", networkPolicy); return networkPolicy; } 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 testRestApiPortNetworkPolicy() { NetworkPolicyPeer clusterOperatorPeer = new NetworkPolicyPeerBuilder() .withNewPodSelector() .withMatchLabels(Collections.singletonMap(Labels.STRIMZI_KIND_LABEL, "cluster-operator")) .endPodSelector() .withNewNamespaceSelector().endNamespaceSelector() .build(); NetworkPolicy np = cc.generateNetworkPolicy(true); assertThat(np.getSpec().getIngress().stream().filter(ing -> ing.getPorts().get(0).getPort().equals(new IntOrString(CruiseControl.REST_API_PORT))).findFirst().orElse(null), is(notNullValue())); List<NetworkPolicyPeer> rules = np.getSpec().getIngress().stream().filter(ing -> ing.getPorts().get(0).getPort().equals(new IntOrString(CruiseControl.REST_API_PORT))).map(NetworkPolicyIngressRule::getFrom).findFirst().orElse(null); assertThat(rules.size(), is(1)); assertThat(rules.contains(clusterOperatorPeer), is(true)); }