private EndpointPort findEndpointPort(EndpointSubset s) {
	List<EndpointPort> ports = s.getPorts();
	EndpointPort endpointPort;
	if (ports.size() == 1) {
		endpointPort = ports.get(0);
	}
	else {
		Predicate<EndpointPort> portPredicate;
		if (!StringUtils.isEmpty(properties.getPrimaryPortName())) {
			portPredicate = port -> properties.getPrimaryPortName()
					.equalsIgnoreCase(port.getName());
		}
		else {
			portPredicate = port -> true;
		}
		endpointPort = ports.stream().filter(portPredicate).findAny()
				.orElseThrow(IllegalStateException::new);
	}
	return endpointPort;
}
 

public static boolean isEndpointsReady(Endpoints e) {
  Utils.checkNotNull(e, "Endpoints can't be null.");
  String name = e.getMetadata().getName();
  Utils.checkNotNull(name, "Endpoints name can't be null.");

  if (e.getSubsets() == null) {
    return false;
  }

  for (EndpointSubset subset : e.getSubsets()) {
    if(!subset.getAddresses().isEmpty() && !subset.getPorts().isEmpty()) {
      return true;
    }
  }
  return false;
}
 

@Scheduled(
		fixedDelayString = "${spring.cloud.kubernetes.discovery.catalogServicesWatchDelay:30000}")
public void catalogServicesWatch() {
	try {
		List<String> previousState = this.catalogEndpointsState.get();

		// not all pods participate in the service discovery. only those that have
		// endpoints.
		List<Endpoints> endpoints = this.properties.isAllNamespaces()
				? this.kubernetesClient.endpoints().inAnyNamespace().list().getItems()
				: this.kubernetesClient.endpoints().list().getItems();
		List<String> endpointsPodNames = endpoints.stream().map(Endpoints::getSubsets)
				.filter(Objects::nonNull).flatMap(Collection::stream)
				.map(EndpointSubset::getAddresses).filter(Objects::nonNull)
				.flatMap(Collection::stream).map(EndpointAddress::getTargetRef)
				.filter(Objects::nonNull).map(ObjectReference::getName) // pod name
																		// unique in
																		// namespace
				.sorted(String::compareTo).collect(Collectors.toList());

		this.catalogEndpointsState.set(endpointsPodNames);

		if (!endpointsPodNames.equals(previousState)) {
			logger.trace("Received endpoints update from kubernetesClient: {}",
					endpointsPodNames);
			this.publisher.publishEvent(new HeartbeatEvent(this, endpointsPodNames));
		}
	}
	catch (Exception e) {
		logger.error("Error watching Kubernetes Services", e);
	}
}
 

private List<ServiceInstance> getNamespaceServiceInstances(EndpointSubsetNS es,
		String serviceId) {
	String namespace = es.getNamespace();
	List<EndpointSubset> subsets = es.getEndpointSubset();
	List<ServiceInstance> instances = new ArrayList<>();
	if (!subsets.isEmpty()) {
		final Service service = this.client.services().inNamespace(namespace)
				.withName(serviceId).get();
		final Map<String, String> serviceMetadata = this.getServiceMetadata(service);
		KubernetesDiscoveryProperties.Metadata metadataProps = this.properties
				.getMetadata();

		for (EndpointSubset s : subsets) {
			// Extend the service metadata map with per-endpoint port information (if
			// requested)
			Map<String, String> endpointMetadata = new HashMap<>(serviceMetadata);
			if (metadataProps.isAddPorts()) {
				Map<String, String> ports = s.getPorts().stream()
						.filter(port -> !StringUtils.isEmpty(port.getName()))
						.collect(toMap(EndpointPort::getName,
								port -> Integer.toString(port.getPort())));
				Map<String, String> portMetadata = getMapWithPrefixedKeys(ports,
						metadataProps.getPortsPrefix());
				if (log.isDebugEnabled()) {
					log.debug("Adding port metadata: " + portMetadata);
				}
				endpointMetadata.putAll(portMetadata);
			}

			List<EndpointAddress> addresses = s.getAddresses();
			for (EndpointAddress endpointAddress : addresses) {
				String instanceId = null;
				if (endpointAddress.getTargetRef() != null) {
					instanceId = endpointAddress.getTargetRef().getUid();
				}

				EndpointPort endpointPort = findEndpointPort(s);
				instances.add(new KubernetesServiceInstance(instanceId, serviceId,
						endpointAddress, endpointPort, endpointMetadata,
						this.isServicePortSecureResolver
								.resolve(new DefaultIsServicePortSecureResolver.Input(
										endpointPort.getPort(),
										service.getMetadata().getName(),
										service.getMetadata().getLabels(),
										service.getMetadata().getAnnotations()))));
			}
		}
	}

	return instances;
}
 

public List<EndpointSubset> getEndpointSubset() {
	return endpointSubset;
}
 

public void setEndpointSubset(List<EndpointSubset> endpointSubset) {
	this.endpointSubset = endpointSubset;
}
 

private List<EndpointSubset> createSubsetsByPodName(String... names) {
	EndpointSubset endpointSubset = new EndpointSubset();
	endpointSubset.setAddresses(createEndpointAddressByPodNames(names));
	return Collections.singletonList(endpointSubset);
}
 

/**
 * Obtains the service port to endpoint address paired map.
 *
 * @param service   kubernetes service
 * @return a map where key is kubernetes service port, and value is the
 * endpoint addresses that are associated with the service port
 */
private Map<ServicePort, Set<String>> getSportEpAddressMap(Service service) {

    Map<ServicePort, Set<String>> map = Maps.newConcurrentMap();

    String serviceName = service.getMetadata().getName();
    List<Endpoints> endpointses = k8sEndpointsService.endpointses()
            .stream()
            .filter(ep -> serviceName.equals(ep.getMetadata().getName()))
            .collect(Collectors.toList());

    service.getSpec().getPorts().stream()
            .filter(Objects::nonNull)
            .filter(sp -> sp.getTargetPort() != null)
            .filter(sp -> sp.getTargetPort().getIntVal() != null ||
                    sp.getTargetPort().getStrVal() != null)
            .forEach(sp -> {
                Integer targetPortInt = sp.getTargetPort().getIntVal() != null ?
                        sp.getTargetPort().getIntVal() : 0;
                String targetPortName = sp.getTargetPort().getStrVal() != null ?
                        sp.getTargetPort().getStrVal() : "";
                String targetProtocol = sp.getProtocol();

                for (Endpoints endpoints : endpointses) {
                    for (EndpointSubset endpointSubset : endpoints.getSubsets()) {

                        // in case service port name is specified but not port number
                        // we will lookup the container port number and use it
                        // as the target port number
                        if (!targetPortName.equals("") && targetPortInt == 0) {
                            for (EndpointAddress addr : endpointSubset.getAddresses()) {
                                Pod pod = podByIp(k8sPodService, addr.getIp());
                                targetPortInt = portNumberByName(pod, targetPortName);
                            }
                        }

                        if (targetPortInt == 0) {
                            continue;
                        }

                        for (EndpointPort endpointPort : endpointSubset.getPorts()) {
                            if (targetProtocol.equals(endpointPort.getProtocol()) &&
                                    (targetPortInt.equals(endpointPort.getPort()) ||
                                            targetPortName.equals(endpointPort.getName()))) {
                                Set<String> addresses = endpointSubset.getAddresses()
                                        .stream().map(EndpointAddress::getIp)
                                        .collect(Collectors.toSet());
                                map.put(sp, addresses);
                            }
                        }
                    }
                }
            });

    return map;
}
 

private void setStatefulGroupFlowRules(DeviceId deviceId, long ctState,
                                       long ctMask, Service service,
                                       boolean install) {
    List<GroupBucket> buckets = Lists.newArrayList();

    String serviceName = service.getMetadata().getName();
    String serviceIp = service.getSpec().getClusterIP();

    // TODO: multi-ports case should be addressed
    Integer servicePort = service.getSpec().getPorts().get(0).getPort();
    String serviceProtocol = service.getSpec().getPorts().get(0).getProtocol();

    String svcStr = servicePortStr(serviceIp, servicePort, serviceProtocol);
    int groupId = svcStr.hashCode();

    List<Endpoints> endpointses = k8sEndpointsService.endpointses()
            .stream()
            .filter(ep -> serviceName.equals(ep.getMetadata().getName()))
            .collect(Collectors.toList());

    Map<String, String> nodeIpGatewayIpMap =
            nodeIpGatewayIpMap(k8sNodeService, k8sNetworkService);

    for (Endpoints endpoints : endpointses) {
        for (EndpointSubset endpointSubset : endpoints.getSubsets()) {
            List<EndpointPort> ports = endpointSubset.getPorts()
                    .stream()
                    .filter(p -> p.getProtocol().equals(TCP))
                    .collect(Collectors.toList());

            for (EndpointAddress address : endpointSubset.getAddresses()) {
                String podIp = nodeIpGatewayIpMap.containsKey(address.getIp()) ?
                        nodeIpGatewayIpMap.get(address.getIp()) : address.getIp();

                NiciraConnTrackTreatmentBuilder connTreatmentBuilder =
                        niciraConnTrackTreatmentBuilder(driverService, deviceId)
                                .commit(true)
                                .natAction(true)
                                .natIp(IpAddress.valueOf(podIp))
                                .natFlag(CT_NAT_DST_FLAG);

                ports.forEach(p -> {
                    ExtensionTreatment ctNatTreatment = connTreatmentBuilder
                            .natPortMin(TpPort.tpPort(p.getPort()))
                            .natPortMax(TpPort.tpPort(p.getPort()))
                            .build();
                    ExtensionTreatment resubmitTreatment = buildResubmitExtension(
                            deviceService.getDevice(deviceId), ACL_TABLE);
                    TrafficTreatment treatment = DefaultTrafficTreatment.builder()
                            .extension(ctNatTreatment, deviceId)
                            .extension(resubmitTreatment, deviceId)
                            .build();
                    buckets.add(buildGroupBucket(treatment, SELECT, (short) -1));
                });
            }
        }
    }

    if (!buckets.isEmpty()) {
        k8sGroupRuleService.setRule(appId, deviceId, groupId, SELECT, buckets, install);

        setTrackNew(deviceId, ctState, ctMask, IpAddress.valueOf(serviceIp),
                TpPort.tpPort(servicePort), NAT_TABLE, groupId,
                PRIORITY_CT_RULE, install);
    }
}