pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaceIsolationPolicy.java - pulsar - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 package org.apache.pulsar.admin.cli;

 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import com.beust.jcommander.Parameters;
 import com.beust.jcommander.converters.CommaParameterSplitter;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.admin.cli.utils.NameValueParameterSplitter;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
 import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
 import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationData;
 import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationDataImpl;
 import org.apache.pulsar.common.policies.data.NamespaceIsolationData;
 import org.apache.pulsar.common.policies.data.NamespaceIsolationDataImpl;

 @Parameters(commandDescription = "Operations about namespace isolation policy")
 public class CmdNamespaceIsolationPolicy extends CmdBase {
     @Parameters(commandDescription = "Create/Update a namespace isolation policy for a cluster. "
             + "This operation requires Pulsar super-user privileges")
     private class SetPolicy extends CliCommand {
         @Parameter(description = "cluster-name policy-name", required = true)
         private List<String> params;

         @Parameter(names = "--namespaces", description = "comma separated namespaces-regex list",
                 required = true, splitter = CommaParameterSplitter.class)
         private List<String> namespaces;

         @Parameter(names = "--primary", description = "comma separated  primary-broker-regex list. "
                 + "In Pulsar, when namespaces (more specifically, namespace bundles) are assigned dynamically to "
                 + "brokers, the namespace isolation policy limits the set of brokers that can be used for assignment. "
                 + "Before topics are assigned to brokers, you can set the namespace isolation policy with a primary or "
                 + "a secondary regex to select desired brokers. If no broker matches the specified regex, you cannot "
                 + "create a topic. If there are not enough primary brokers, topics are assigned to secondary brokers. "
                 + "If there are not enough secondary brokers, topics are assigned to other brokers which do not have "
                 + "any isolation policies.", required = true, splitter = CommaParameterSplitter.class)
         private List<String> primary;

         @Parameter(names = "--secondary", description = "comma separated secondary-broker-regex list",
                 required = false, splitter = CommaParameterSplitter.class)
         private List<String> secondary = new ArrayList<String>(); // optional

         @Parameter(names = "--auto-failover-policy-type",
                 description = "auto failover policy type name ['min_available']", required = true)
         private String autoFailoverPolicyTypeName;

         @Parameter(names = "--auto-failover-policy-params",
                 description = "comma separated name=value auto failover policy parameters",
                 required = true, converter = NameValueParameterSplitter.class)
         private Map<String, String> autoFailoverPolicyParams;

         void run() throws PulsarAdminException {
             String clusterName = getOneArgument(params, 0, 2);
             String policyName = getOneArgument(params, 1, 2);

             // validate and create the POJO
             NamespaceIsolationData namespaceIsolationData = createNamespaceIsolationData(namespaces, primary, secondary,
                     autoFailoverPolicyTypeName, autoFailoverPolicyParams);

             getAdmin().clusters().createNamespaceIsolationPolicy(clusterName, policyName, namespaceIsolationData);
         }
     }

     @Parameters(commandDescription = "List all namespace isolation policies of a cluster. "
             + "This operation requires Pulsar super-user privileges")
     private class GetAllPolicies extends CliCommand {
         @Parameter(description = "cluster-name", required = true)
         private List<String> params;

         void run() throws PulsarAdminException {
             String clusterName = getOneArgument(params);

             Map<String, ? extends NamespaceIsolationData> policyMap =
                     getAdmin().clusters().getNamespaceIsolationPolicies(clusterName);

             print(policyMap);
         }
     }

     @Parameters(commandDescription = "List all brokers with namespace-isolation policies attached to it. "
             + "This operation requires Pulsar super-user privileges")
     private class GetAllBrokersWithPolicies extends CliCommand {
         @Parameter(description = "cluster-name", required = true)
         private List<String> params;

         void run() throws PulsarAdminException {
             String clusterName = getOneArgument(params);

             List<BrokerNamespaceIsolationData> brokers = getAdmin().clusters()
                     .getBrokersWithNamespaceIsolationPolicy(clusterName);
             List<BrokerNamespaceIsolationDataImpl> data = new ArrayList<>();
             brokers.forEach(v -> data.add((BrokerNamespaceIsolationDataImpl) v));
             print(data);
         }
     }

     @Parameters(commandDescription = "Get broker with namespace-isolation policies attached to it. "
             + "This operation requires Pulsar super-user privileges")
     private class GetBrokerWithPolicies extends CliCommand {
         @Parameter(description = "cluster-name", required = true)
         private List<String> params;

         @Parameter(names = "--broker",
                 description = "Broker-name to get namespace-isolation policies attached to it", required = true)
         private String broker;

         void run() throws PulsarAdminException {
             String clusterName = getOneArgument(params);

             BrokerNamespaceIsolationDataImpl brokerData = (BrokerNamespaceIsolationDataImpl) getAdmin().clusters()
                     .getBrokerWithNamespaceIsolationPolicy(clusterName, broker);

             print(brokerData);
         }
     }

     @Parameters(commandDescription = "Get namespace isolation policy of a cluster. "
             + "This operation requires Pulsar super-user privileges")
     private class GetPolicy extends CliCommand {
         @Parameter(description = "cluster-name policy-name", required = true)
         private List<String> params;

         void run() throws PulsarAdminException {
             String clusterName = getOneArgument(params, 0, 2);
             String policyName = getOneArgument(params, 1, 2);

             NamespaceIsolationDataImpl nsIsolationData = (NamespaceIsolationDataImpl) getAdmin().clusters()
                     .getNamespaceIsolationPolicy(clusterName, policyName);

             print(nsIsolationData);
         }
     }

     @Parameters(commandDescription = "Delete namespace isolation policy of a cluster. "
             + "This operation requires Pulsar super-user privileges")
     private class DeletePolicy extends CliCommand {
         @Parameter(description = "cluster-name policy-name", required = true)
         private List<String> params;

         void run() throws PulsarAdminException {
             String clusterName = getOneArgument(params, 0, 2);
             String policyName = getOneArgument(params, 1, 2);

             getAdmin().clusters().deleteNamespaceIsolationPolicy(clusterName, policyName);
         }
     }

     private List<String> validateList(List<String> list) {
         return list.stream()
                 .filter(StringUtils::isNotEmpty)
                 .collect(Collectors.toList());
     }

     private NamespaceIsolationData createNamespaceIsolationData(List<String> namespaces,
                                                                 List<String> primary,
                                                                 List<String> secondary,
                                                                 String autoFailoverPolicyTypeName,
                                                                 Map<String, String> autoFailoverPolicyParams) {

         // validate
         namespaces = validateList(namespaces);
         if (namespaces.isEmpty()) {
             throw new ParameterException("unable to parse namespaces parameter list: " + namespaces);
         }

         primary = validateList(primary);
         if (primary.isEmpty()) {
             throw new ParameterException("unable to parse primary parameter list: " + namespaces);
         }

         secondary = validateList(secondary);

         // System.out.println("namespaces = " + namespaces);
         // System.out.println("primary = " + primary);
         // System.out.println("secondary = " + secondary);
         // System.out.println("autoFailoverPolicyTypeName = " + autoFailoverPolicyTypeName);
         // System.out.println("autoFailoverPolicyParams = " + autoFailoverPolicyParams);

         NamespaceIsolationData.Builder nsIsolationDataBuilder = NamespaceIsolationData.builder();

         if (namespaces != null) {
             nsIsolationDataBuilder.namespaces(namespaces);
         }

         if (primary != null) {
             nsIsolationDataBuilder.primary(primary);
         }

         if (secondary != null) {
             nsIsolationDataBuilder.secondary(secondary);
         }

         AutoFailoverPolicyType policyType = AutoFailoverPolicyType.fromString(autoFailoverPolicyTypeName);

         nsIsolationDataBuilder.autoFailoverPolicy(AutoFailoverPolicyData.builder()
                 .policyType(policyType)
                 .parameters(autoFailoverPolicyParams)
                 .build());

         // validation if necessary
         if (policyType == AutoFailoverPolicyType.min_available) {
             // ignore
             boolean error = true;
             String[] expectParamKeys = { "min_limit", "usage_threshold" };

             if (autoFailoverPolicyParams.size() == expectParamKeys.length) {
                 for (String paramKey : expectParamKeys) {
                     if (!autoFailoverPolicyParams.containsKey(paramKey)) {
                         break;
                     }
                 }
                 error = false;
             }

             if (error) {
                 throw new ParameterException(
                         "Unknown auto failover policy params specified : " + autoFailoverPolicyParams);
             }

         } else {
             // either we don't handle the new type or user has specified a bad type
             throw new ParameterException("Unknown auto failover policy type specified : " + autoFailoverPolicyTypeName);
         }

         return nsIsolationDataBuilder.build();
     }

     public CmdNamespaceIsolationPolicy(Supplier<PulsarAdmin> admin) {
         super("ns-isolation-policy", admin);
         jcommander.addCommand("set", new SetPolicy());
         jcommander.addCommand("get", new GetPolicy());
         jcommander.addCommand("list", new GetAllPolicies());
         jcommander.addCommand("delete", new DeletePolicy());
         jcommander.addCommand("brokers", new GetAllBrokersWithPolicies());
         jcommander.addCommand("broker", new GetBrokerWithPolicies());
     }

 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	package org.apache.pulsar.admin.cli;

	import com.beust.jcommander.Parameter;
	import com.beust.jcommander.ParameterException;
	import com.beust.jcommander.Parameters;
	import com.beust.jcommander.converters.CommaParameterSplitter;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.Map;
	import java.util.function.Supplier;
	import java.util.stream.Collectors;
	import org.apache.commons.lang3.StringUtils;
	import org.apache.pulsar.admin.cli.utils.NameValueParameterSplitter;
	import org.apache.pulsar.client.admin.PulsarAdmin;
	import org.apache.pulsar.client.admin.PulsarAdminException;
	import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
	import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
	import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationData;
	import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationDataImpl;
	import org.apache.pulsar.common.policies.data.NamespaceIsolationData;
	import org.apache.pulsar.common.policies.data.NamespaceIsolationDataImpl;

	@Parameters(commandDescription = "Operations about namespace isolation policy")
	public class CmdNamespaceIsolationPolicy extends CmdBase {
	@Parameters(commandDescription = "Create/Update a namespace isolation policy for a cluster. "
	+ "This operation requires Pulsar super-user privileges")
	private class SetPolicy extends CliCommand {
	@Parameter(description = "cluster-name policy-name", required = true)
	private List<String> params;

	@Parameter(names = "--namespaces", description = "comma separated namespaces-regex list",
	required = true, splitter = CommaParameterSplitter.class)
	private List<String> namespaces;

	@Parameter(names = "--primary", description = "comma separated primary-broker-regex list. "
	+ "In Pulsar, when namespaces (more specifically, namespace bundles) are assigned dynamically to "
	+ "brokers, the namespace isolation policy limits the set of brokers that can be used for assignment. "
	+ "Before topics are assigned to brokers, you can set the namespace isolation policy with a primary or "
	+ "a secondary regex to select desired brokers. If no broker matches the specified regex, you cannot "
	+ "create a topic. If there are not enough primary brokers, topics are assigned to secondary brokers. "
	+ "If there are not enough secondary brokers, topics are assigned to other brokers which do not have "
	+ "any isolation policies.", required = true, splitter = CommaParameterSplitter.class)
	private List<String> primary;

	@Parameter(names = "--secondary", description = "comma separated secondary-broker-regex list",
	required = false, splitter = CommaParameterSplitter.class)
	private List<String> secondary = new ArrayList<String>(); // optional

	@Parameter(names = "--auto-failover-policy-type",
	description = "auto failover policy type name ['min_available']", required = true)
	private String autoFailoverPolicyTypeName;

	@Parameter(names = "--auto-failover-policy-params",
	description = "comma separated name=value auto failover policy parameters",
	required = true, converter = NameValueParameterSplitter.class)
	private Map<String, String> autoFailoverPolicyParams;

	void run() throws PulsarAdminException {
	String clusterName = getOneArgument(params, 0, 2);
	String policyName = getOneArgument(params, 1, 2);

	// validate and create the POJO
	NamespaceIsolationData namespaceIsolationData = createNamespaceIsolationData(namespaces, primary, secondary,
	autoFailoverPolicyTypeName, autoFailoverPolicyParams);

	getAdmin().clusters().createNamespaceIsolationPolicy(clusterName, policyName, namespaceIsolationData);
	}
	}

	@Parameters(commandDescription = "List all namespace isolation policies of a cluster. "
	+ "This operation requires Pulsar super-user privileges")
	private class GetAllPolicies extends CliCommand {
	@Parameter(description = "cluster-name", required = true)
	private List<String> params;

	void run() throws PulsarAdminException {
	String clusterName = getOneArgument(params);

	Map<String, ? extends NamespaceIsolationData> policyMap =
	getAdmin().clusters().getNamespaceIsolationPolicies(clusterName);

	print(policyMap);
	}
	}

	@Parameters(commandDescription = "List all brokers with namespace-isolation policies attached to it. "
	+ "This operation requires Pulsar super-user privileges")
	private class GetAllBrokersWithPolicies extends CliCommand {
	@Parameter(description = "cluster-name", required = true)
	private List<String> params;

	void run() throws PulsarAdminException {
	String clusterName = getOneArgument(params);

	List<BrokerNamespaceIsolationData> brokers = getAdmin().clusters()
	.getBrokersWithNamespaceIsolationPolicy(clusterName);
	List<BrokerNamespaceIsolationDataImpl> data = new ArrayList<>();
	brokers.forEach(v -> data.add((BrokerNamespaceIsolationDataImpl) v));
	print(data);
	}
	}

	@Parameters(commandDescription = "Get broker with namespace-isolation policies attached to it. "
	+ "This operation requires Pulsar super-user privileges")
	private class GetBrokerWithPolicies extends CliCommand {
	@Parameter(description = "cluster-name", required = true)
	private List<String> params;

	@Parameter(names = "--broker",
	description = "Broker-name to get namespace-isolation policies attached to it", required = true)
	private String broker;

	void run() throws PulsarAdminException {
	String clusterName = getOneArgument(params);

	BrokerNamespaceIsolationDataImpl brokerData = (BrokerNamespaceIsolationDataImpl) getAdmin().clusters()
	.getBrokerWithNamespaceIsolationPolicy(clusterName, broker);

	print(brokerData);
	}
	}

	@Parameters(commandDescription = "Get namespace isolation policy of a cluster. "
	+ "This operation requires Pulsar super-user privileges")
	private class GetPolicy extends CliCommand {
	@Parameter(description = "cluster-name policy-name", required = true)
	private List<String> params;

	void run() throws PulsarAdminException {
	String clusterName = getOneArgument(params, 0, 2);
	String policyName = getOneArgument(params, 1, 2);

	NamespaceIsolationDataImpl nsIsolationData = (NamespaceIsolationDataImpl) getAdmin().clusters()
	.getNamespaceIsolationPolicy(clusterName, policyName);

	print(nsIsolationData);
	}
	}

	@Parameters(commandDescription = "Delete namespace isolation policy of a cluster. "
	+ "This operation requires Pulsar super-user privileges")
	private class DeletePolicy extends CliCommand {
	@Parameter(description = "cluster-name policy-name", required = true)
	private List<String> params;

	void run() throws PulsarAdminException {
	String clusterName = getOneArgument(params, 0, 2);
	String policyName = getOneArgument(params, 1, 2);

	getAdmin().clusters().deleteNamespaceIsolationPolicy(clusterName, policyName);
	}
	}

	private List<String> validateList(List<String> list) {
	return list.stream()
	.filter(StringUtils::isNotEmpty)
	.collect(Collectors.toList());
	}

	private NamespaceIsolationData createNamespaceIsolationData(List<String> namespaces,
	List<String> primary,
	List<String> secondary,
	String autoFailoverPolicyTypeName,
	Map<String, String> autoFailoverPolicyParams) {

	// validate
	namespaces = validateList(namespaces);
	if (namespaces.isEmpty()) {
	throw new ParameterException("unable to parse namespaces parameter list: " + namespaces);
	}

	primary = validateList(primary);
	if (primary.isEmpty()) {
	throw new ParameterException("unable to parse primary parameter list: " + namespaces);
	}

	secondary = validateList(secondary);

	// System.out.println("namespaces = " + namespaces);
	// System.out.println("primary = " + primary);
	// System.out.println("secondary = " + secondary);
	// System.out.println("autoFailoverPolicyTypeName = " + autoFailoverPolicyTypeName);
	// System.out.println("autoFailoverPolicyParams = " + autoFailoverPolicyParams);

	NamespaceIsolationData.Builder nsIsolationDataBuilder = NamespaceIsolationData.builder();

	if (namespaces != null) {
	nsIsolationDataBuilder.namespaces(namespaces);
	}

	if (primary != null) {
	nsIsolationDataBuilder.primary(primary);
	}

	if (secondary != null) {
	nsIsolationDataBuilder.secondary(secondary);
	}

	AutoFailoverPolicyType policyType = AutoFailoverPolicyType.fromString(autoFailoverPolicyTypeName);

	nsIsolationDataBuilder.autoFailoverPolicy(AutoFailoverPolicyData.builder()
	.policyType(policyType)
	.parameters(autoFailoverPolicyParams)
	.build());

	// validation if necessary
	if (policyType == AutoFailoverPolicyType.min_available) {
	// ignore
	boolean error = true;
	String[] expectParamKeys = { "min_limit", "usage_threshold" };

	if (autoFailoverPolicyParams.size() == expectParamKeys.length) {
	for (String paramKey : expectParamKeys) {
	if (!autoFailoverPolicyParams.containsKey(paramKey)) {
	break;
	}
	}
	error = false;
	}

	if (error) {
	throw new ParameterException(
	"Unknown auto failover policy params specified : " + autoFailoverPolicyParams);
	}

	} else {
	// either we don't handle the new type or user has specified a bad type
	throw new ParameterException("Unknown auto failover policy type specified : " + autoFailoverPolicyTypeName);
	}

	return nsIsolationDataBuilder.build();
	}

	public CmdNamespaceIsolationPolicy(Supplier<PulsarAdmin> admin) {
	super("ns-isolation-policy", admin);
	jcommander.addCommand("set", new SetPolicy());
	jcommander.addCommand("get", new GetPolicy());
	jcommander.addCommand("list", new GetAllPolicies());
	jcommander.addCommand("delete", new DeletePolicy());
	jcommander.addCommand("brokers", new GetAllBrokersWithPolicies());
	jcommander.addCommand("broker", new GetBrokerWithPolicies());
	}

	}