import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
@Test
public void testClusterIsCreated() throws Exception {
    Address queue = createAddress("myqueue", StandardType.QUEUE);
    KubernetesList resources = new KubernetesList();
    resources.setItems(Arrays.asList(new ConfigMap()));
    AddressCluster cluster = new AddressCluster("myqueue", resources);

    when(mockHelper.listClusters()).thenReturn(Collections.emptyList());
    when(mockGenerator.generateCluster("myqueue", Collections.singleton(queue))).thenReturn(cluster);
    ArgumentCaptor<Set<io.enmasse.address.model.Address>> arg = ArgumentCaptor.forClass(Set.class);

    controller.resourcesUpdated(Collections.singleton(queue));
    verify(mockGenerator).generateCluster(eq("myqueue"), arg.capture());
    assertThat(arg.getValue(), hasItem(queue));
    verify(mockHelper).create(resources);
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
@Test
public void testNodesAreRetained() throws Exception {
    Address queue = createAddress("myqueue", StandardType.QUEUE);

    KubernetesList resources = new KubernetesList();
    resources.setItems(Arrays.asList(new ConfigMap()));
    AddressCluster existing = new AddressCluster(queue.getAddress(), resources);
    when(mockHelper.listClusters()).thenReturn(Collections.singletonList(existing));

    Address newQueue = createAddress("newqueue", StandardType.QUEUE);
    AddressCluster newCluster = new AddressCluster(newQueue.getAddress(), resources);

    when(mockGenerator.generateCluster("newqueue", Collections.singleton(newQueue))).thenReturn(newCluster);
    ArgumentCaptor<Set<io.enmasse.address.model.Address>> arg = ArgumentCaptor.forClass(Set.class);

    controller.resourcesUpdated(Sets.newSet(queue, newQueue));

    verify(mockGenerator).generateCluster(anyString(), arg.capture());
    assertThat(arg.getValue(), is(Sets.newSet(newQueue)));
    verify(mockHelper).create(resources);
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
@Test
public void testAddressesAreGrouped() throws Exception {
    Address addr0 = createAddress("myqueue0", StandardType.QUEUE);
    Address addr1 = createAddress("myqueue1", StandardType.QUEUE, "pooled-inmemory");
    Address addr2 = createAddress("myqueue2", StandardType.QUEUE, "pooled-inmemory");
    Address addr3 = createAddress("myqueue3", StandardType.QUEUE);

    KubernetesList resources = new KubernetesList();
    resources.setItems(Arrays.asList(new ConfigMap()));
    AddressCluster existing = new AddressCluster("myqueue0", resources);

    when(mockHelper.listClusters()).thenReturn(Collections.singletonList(existing));
    ArgumentCaptor<Set<io.enmasse.address.model.Address>> arg = ArgumentCaptor.forClass(Set.class);
    when(mockGenerator.generateCluster(anyString(), arg.capture())).thenReturn(new AddressCluster("foo", resources));

    controller.resourcesUpdated(Sets.newSet(addr0, addr1, addr2, addr3));

    Set<io.enmasse.address.model.Address> generated = arg.getAllValues().stream().flatMap(Collection::stream).collect(Collectors.toSet());
    assertThat(generated.size(), is(3));
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
@Test
public void testAddressesAreNotRecreated() throws Exception {
    Address address = createAddress("addr1", StandardType.ANYCAST);
    Address newAddress = createAddress("addr2", StandardType.ANYCAST);

    KubernetesList resources = new KubernetesList();
    when(mockGenerator.generateCluster(eq("addr1"), anySet())).thenReturn(new AddressCluster("addr1", resources));
    when(mockGenerator.generateCluster(eq("addr2"), anySet())).thenReturn(new AddressCluster("addr2", resources));

    doThrow(new KubernetesClientException("Unable to replace resource")).when(mockApi).replaceAddress(address);

    try {
        controller.resourcesUpdated(Sets.newSet(address, newAddress));

        ArgumentCaptor<Address> addressArgumentCaptor = ArgumentCaptor.forClass(Address.class);
        verify(mockApi, times(2)).replaceAddress(addressArgumentCaptor.capture());
        List<Address> replaced = addressArgumentCaptor.getAllValues();
        assertThat(replaced, hasItem(newAddress));
    } catch (KubernetesClientException e) {
        fail("Should not throw exception with multiple items");
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
private void createFromUri(String prjName, String uri) throws OpenShiftClientException {
    URL url = toUrl(uri);
    if (url != null) {
        KubernetesList kubeList = delegate.lists().load(url).get();
        List<HasMetadata> items = kubeList.getItems();
        if (items.size() > 0) {
            for (HasMetadata item : items) {
                String name = item.getMetadata().getName();
                if (item instanceof ServiceAccount) {
                    if (delegate.serviceAccounts().inNamespace(prjName).withName(name).get() == null) {
                        setGuvnorAlaGenerated(item);
                    }
                } else if (item instanceof Secret) {
                    if (delegate.secrets().inNamespace(prjName).withName(name).get() == null) {
                        setGuvnorAlaGenerated(item);
                    }
                } else if (item instanceof ImageStream) {
                    if (delegate.imageStreams().inNamespace(prjName).withName(name).get() == null) {
                        setGuvnorAlaGenerated(item);
                    }
                }
            }
            delegate.lists().inNamespace(prjName).create(kubeList);
        }
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
private void createFromTemplate(OpenShiftRuntimeConfig runtimeConfig) throws OpenShiftClientException {
    OpenShiftTemplate template = new OpenShiftTemplate(this, runtimeConfig);
    Map<String, String> parameters = new LinkedHashMap<String, String>();
    parameters.putAll(OpenShiftParameters.fromRuntimeConfig(runtimeConfig));
    String kieServerContainerDeployment = runtimeConfig.getKieServerContainerDeployment();
    if (kieServerContainerDeployment != null && !kieServerContainerDeployment.trim().isEmpty()) {
        parameters.put(OpenShiftProperty.KIE_SERVER_CONTAINER_DEPLOYMENT.envKey(), kieServerContainerDeployment);
    }
    KubernetesList kubeList = template.process(parameters);
    if (kubeList != null && kubeList.getItems().size() > 0) {
        try {
            DeploymentConfig dc = getDeploymentConfig(kubeList, runtimeConfig.getServiceName());
            if (dc != null) {
                dc.getSpec().setReplicas(0);
            }
            String prjName = runtimeConfig.getProjectName();
            delegate.lists().inNamespace(prjName).create(kubeList);
        } catch (Throwable t) {
            throw new OpenShiftClientException(t.getMessage(), t);
        }
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
public OpenShiftTemplate(OpenShiftClient client, OpenShiftRuntimeConfig runtimeConfig) {
    Loadable<TemplateResource<Template, KubernetesList, DoneableTemplate>> loadable = null;
    String prjName = runtimeConfig.getProjectName();
    String templateUri = runtimeConfig.getResourceTemplateUri();
    String templateName = runtimeConfig.getResourceTemplateName();
    if (prjName != null) {
        loadable = client.getDelegate().templates().inNamespace(prjName);
    } else {
        loadable = client.getDelegate().templates();
    }
    TemplateResource<Template, KubernetesList, DoneableTemplate> res = null;
    if (templateUri != null) {
        URL templateUrl = client.toUrl(templateUri);
        if (templateUrl != null) {
            res = loadable.load(templateUrl);
        }
    } else if (templateName != null && !templateName.isEmpty()) {
        res = loadable.load(templateName);
    }
    resource = res;
    template = res != null ? res.get() : null;
    if (template == null) {
        throw new OpenShiftClientException(String.format("could not load template with project [%s] and uri [%s] or name [%s]", prjName, templateUri, templateName));
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
public KubernetesList processTemplate(Template template, Map<String, String> parameters) {
	ParameterValue[] values = processParameters(parameters);

	return withDefaultUser(client -> {
		client.templates().createOrReplace(template);

		return client.templates().withName(template.getMetadata().getName()).process(values);
	});
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
public static void deploy() {
	Template template = null;
	try (InputStream is = ClassLoader.getSystemResourceAsStream("templates/syndesis-amq.yml")) {
		template = OpenShiftUtils.getInstance().withDefaultUser(client -> client.templates().load(is).get());
	} catch (IOException ex) {
		throw new IllegalArgumentException("Unable to read template ", ex);
	}
	Map<String, String> templateParams = new HashMap<>();
	templateParams.put("MQ_USERNAME", "amq");
	templateParams.put("MQ_PASSWORD", "topSecret");

	// try to clean previous broker
	cleanUp();
	OpenShiftUtils.getInstance().withDefaultUser(c -> c.templates().withName("syndesis-amq").delete());

	KubernetesList processedTemplate = OpenShiftUtils.getInstance().processTemplate(template, templateParams);
	OpenShiftUtils.getInstance().createResources(processedTemplate);

	try {
		OpenShiftWaitUtils.waitFor(OpenShiftWaitUtils.isAPodReady("application", "broker"));
	} catch (InterruptedException | TimeoutException e) {
		log.error("Wait for syndesis-rest failed ", e);
	}
	Account amqAccount = new Account();
	amqAccount.setService("amq");
	Map<String, String> accountParameters = new HashMap<>();
	accountParameters.put("brokerUrl", "tcp://broker-amq:61616");
	accountParameters.put("username", "amq");
	accountParameters.put("password", "topSecret");
	accountParameters.put("clientId", UUID.randomUUID().toString());
	amqAccount.setProperties(accountParameters);
	AccountsDirectory.getInstance().addAccount("AMQ", amqAccount);
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
@Override
public KubernetesList processTemplate(String templateName, ParameterValue... parameterValues) {
    if (templateDir != null) {
        File templateFile = new File(templateDir, templateName + TEMPLATE_SUFFIX);
        return client.templates().load(templateFile).processLocally(parameterValues);
    } else {
        return client.templates().withName(templateName).process(parameterValues);
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
/**
 * Generate cluster for a given destination group.
 *
 * NOTE: This method assumes that all destinations within a group share the same properties.
 *
 * @param addressSet The set of addresses to generate cluster for
 */
public AddressCluster generateCluster(String clusterId, Set<Address> addressSet) {
    Address first = addressSet.iterator().next();

    Plan plan = addressResolver.getPlan(first);
    KubernetesList resources = plan.getTemplateConfig().map(t -> processTemplate(clusterId, first, addressSet, t)).orElse(new KubernetesList());
    return new AddressCluster(clusterId, resources);
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
private KubernetesList processTemplate(String clusterId, Address first, Set<Address> addressSet, TemplateConfig templateConfig) {
    Map<String, String> paramMap = new LinkedHashMap<>(templateConfig.getParameters());

    paramMap.put(TemplateParameter.NAME, KubeUtil.sanitizeName(clusterId));
    paramMap.put(TemplateParameter.CLUSTER_ID, clusterId);
    paramMap.put(TemplateParameter.ADDRESS_SPACE, first.getAddressSpace());
    paramMap.put(TemplateParameter.COLOCATED_ROUTER_SECRET, templateOptions.getMessagingSecret());
    paramMap.put(TemplateParameter.AUTHENTICATION_SERVICE_HOST, templateOptions.getAuthenticationServiceHost());
    paramMap.put(TemplateParameter.AUTHENTICATION_SERVICE_PORT, templateOptions.getAuthenticationServicePort());
    paramMap.put(TemplateParameter.AUTHENTICATION_SERVICE_CA_SECRET, templateOptions.getAuthenticationServiceCaSecret());
    paramMap.put(TemplateParameter.AUTHENTICATION_SERVICE_CLIENT_SECRET, templateOptions.getAuthenticationServiceClientSecret());
    paramMap.put(TemplateParameter.AUTHENTICATION_SERVICE_SASL_INIT_HOST, templateOptions.getAuthenticationServiceSaslInitHost());

    // If the name of the group matches that of the address, assume a scalable queue
    if (clusterId.equals(first.getName()) && addressSet.size() == 1) {
        paramMap.put(TemplateParameter.ADDRESS, first.getAddress());
    }

    ParameterValue parameters[] = paramMap.entrySet().stream()
            .map(entry -> new ParameterValue(entry.getKey(), entry.getValue()))
            .collect(Collectors.toList())
            .toArray(new ParameterValue[0]);


    KubernetesList items = kubernetes.processTemplate(templateConfig.getName(), parameters);


    // These are attributes that we need to identify components belonging to this address
    Kubernetes.addObjectAnnotation(items, AnnotationKeys.CLUSTER_ID, clusterId);
    Kubernetes.addObjectAnnotation(items, AnnotationKeys.ADDRESS_SPACE, first.getAddressSpace());
    Kubernetes.addObjectLabel(items, LabelKeys.UUID, first.getUuid());
    return items;
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
static void addObjectLabel(KubernetesList items, String labelKey, String labelValue) {
    for (HasMetadata item : items.getItems()) {
        Map<String, String> labels = item.getMetadata().getLabels();
        if (labels == null) {
            labels = new LinkedHashMap<>();
        }
        labels.put(labelKey, labelValue);
        item.getMetadata().setLabels(labels);
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
static void addObjectAnnotation(KubernetesList items, String annotationKey, String annotationValue) {
    for (HasMetadata item : items.getItems()) {
        Map<String, String> annotations = item.getMetadata().getAnnotations();
        if (annotations == null) {
            annotations = new LinkedHashMap<>();
        }
        annotations.put(annotationKey, annotationValue);
        item.getMetadata().setAnnotations(annotations);
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
KubernetesList process(Map<String, String> parameters) {
    if (resource != null) {
        return resource.process(parameters);
    } else {
        throw new IllegalStateException("cannot process parameters with null template resource");
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
@Test
public void createCheServer() throws Exception {
    OpenShiftClient openShiftClient = null;
    try {
        openShiftClient = client.get(endpoint, username, password);

        ProjectRequest projectRequest = createTestProject(openShiftClient);
        LOG.info("Number of projects: {}", getNumberOfProjects(openShiftClient));

        LOG.info("Test project has been deleted: {}", deleteTestProject(openShiftClient, projectRequest));
        LOG.info("Number of projects: {}", getNumberOfProjects(openShiftClient));

        // Controller controller = new Controller(client);
        // controller.applyJson(template.get());

        Template template = loadTemplate(openShiftClient);

        List<Parameter> parameters = template.getParameters();
        LOG.info("Number of template parameters: {}", parameters.size());

        List<ParameterValue> pvs = new ArrayList<>();
        for (Parameter parameter : parameters) {
            String name = parameter.getName();
            String value = parameter.getValue();
            LOG.info("Template Parameter Name: {}", name);
            LOG.info("Template Parameter Value: {}", value);
            if (CHE_OPENSHIFT_ENDPOINT.equals(name) && value.isEmpty()) {
                value = endpoint;
            }
            pvs.add(new ParameterValue(name, value));
        }

        KubernetesList list = processTemplate(openShiftClient, pvs);
        createResources(openShiftClient, list);

        Pod pod = openShiftClient.pods().inNamespace(project).withName("che-host").get();

        Route route = openShiftClient.routes().inNamespace(project).withName("che-host").get();

        LOG.info("Pods: {}", getNumberOfPods(openShiftClient));
    } finally {
        if (openShiftClient != null) {
            openShiftClient.close();
        }
    }
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
private KubernetesList processTemplate(OpenShiftClient client, List<ParameterValue> parameterValues)
        throws IOException {
    ClientTemplateResource<Template, KubernetesList, DoneableTemplate> templateHandle = client.templates()
            .load(cheServerTemplate.getInputStream());
    return templateHandle.process(parameterValues.toArray(new ParameterValue[parameterValues.size()]));
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
private KubernetesList createResources(OpenShiftClient client, KubernetesList list) {
    return client.lists().inNamespace(project).create(list);
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
public KubernetesList createResources(KubernetesList list) {
	return (KubernetesList) this.withDefaultUser((client) -> {
		return (KubernetesList) client.lists().create(new KubernetesList[]{list});
	});
}
 

import io.fabric8.kubernetes.api.model.KubernetesList; //导入依赖的package包/类
public static void deploy() {
	OpenShiftUtils.getInstance().cleanProject();

	// get & create restricted SA
	ServiceAccount serviceAccount1 = OpenShiftUtils.getInstance().withDefaultUser(client -> client.serviceAccounts().createOrReplace(getSupportSA()));
	// get token from SA `oc secrets get-token` && wait until created to prevent 404
	TestUtils.waitForEvent(Optional::isPresent,
			() -> OpenShiftUtils.getInstance().getSecrets().stream().filter(s -> s.getMetadata().getName().startsWith("syndesis-oauth-client-token")).findFirst(),
			TimeUnit.MINUTES,
			2,
			TimeUnit.SECONDS,
			5);

	Secret secret = OpenShiftUtils.getInstance().getSecrets().stream().filter(s -> s.getMetadata().getName().startsWith("syndesis-oauth-client-token")).findFirst().get();
	// token is Base64 encoded by default
	String oauthTokenEncoded = secret.getData().get("token");
	byte[] oauthTokenBytes = Base64.decodeBase64(oauthTokenEncoded);
	String oauthToken = new String(oauthTokenBytes);

	// get the template
	Template template = getTemplate();
	// set params
	Map<String, String> templateParams = new HashMap<>();
	templateParams.put("ROUTE_HOSTNAME", TestConfiguration.openShiftNamespace() + "." + TestConfiguration.syndesisUrlSuffix());
	templateParams.put("OPENSHIFT_MASTER", TestConfiguration.openShiftUrl());
	templateParams.put("OPENSHIFT_PROJECT", TestConfiguration.openShiftNamespace());
	templateParams.put("OPENSHIFT_OAUTH_CLIENT_SECRET", oauthToken);
	templateParams.put("TEST_SUPPORT_ENABLED", "true");
	// process & create
	KubernetesList processedTemplate = OpenShiftUtils.getInstance().processTemplate(template, templateParams);
	OpenShiftUtils.getInstance().createResources(processedTemplate);
	OpenShiftUtils.getInstance().createRestRoute();

	//TODO: there's a bug in openshift-client, we need to initialize manually
	OpenShiftUtils.getInstance().withDefaultUser(client -> client.roleBindings().createOrReplaceWithNew()
			.withNewMetadata()
				.withName("syndesis:editors")
	        .endMetadata()
			.withNewRoleRef().withName("edit").endRoleRef()
			.addNewSubject().withKind("ServiceAccount").withName("syndesis-rest").withNamespace(TestConfiguration.openShiftNamespace()).endSubject()
			.addToUserNames(String.format("system:serviceaccount:%s:syndesis-rest", TestConfiguration.openShiftNamespace()))
			.done()
	);
}