Add java API to create cluster with CloudConfig

helix-core/src/main/java/org/apache/helix/ConfigAccessor.java

@@ -591,7 +591,7 @@ public ClusterConfig getClusterConfig(String clusterName) {
    * @return The instance of {@link CloudConfig}
    */
   public CloudConfig getCloudConfig(String clusterName) {
-    if (!ZKUtil.isClusterSetup(clusterName, zkClient)) {
+    if (!ZKUtil.isClusterSetup(clusterName, _zkClient)) {
       throw new HelixException(
           String.format("Failed to get config. cluster: %s is not setup.", clusterName));
     }
@@ -604,7 +604,7 @@ public CloudConfig getCloudConfig(String clusterName) {
       return null;
     }
 
-    return new CloudConfig(record);
+    return new CloudConfig.Builder(record).build();
   }
 
   /**

helix-core/src/main/java/org/apache/helix/HelixAdmin.java

@@ -22,7 +22,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
-
+import org.apache.helix.model.CloudConfig;
 import org.apache.helix.model.ClusterConstraints;
 import org.apache.helix.model.ClusterConstraints.ConstraintType;
 import org.apache.helix.model.ConstraintItem;
@@ -380,6 +380,19 @@ void addStateModelDef(String clusterName, String stateModelDef, StateModelDefini
    */
   void dropResource(String clusterName, String resourceName);
 
+  /**
+   * Add cloud config to the cluster.
+   * @param clusterName
+   * @param cloudConfig
+   */
+  void addCloudConfig(String clusterName, CloudConfig cloudConfig);
+
+  /**
+   * Remove the Cloud Config for specific cluster
+   * @param clusterName
+   */
+  void removeCloudConfig(String clusterName);
+
   /**
    * Get a list of state model definitions in a cluster
    * @param clusterName

helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java

@@ -59,6 +59,7 @@
 import org.apache.helix.controller.rebalancer.strategy.RebalanceStrategy;
 import org.apache.helix.manager.zk.client.HelixZkClient;
 import org.apache.helix.manager.zk.client.SharedZkClientFactory;
+import org.apache.helix.model.CloudConfig;
 import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.ClusterConstraints;
 import org.apache.helix.model.ClusterConstraints.ConstraintType;
@@ -1025,6 +1026,33 @@ public void dropResource(String clusterName, String resourceName) {
     accessor.removeProperty(keyBuilder.resourceConfig(resourceName));
   }
 
+  @Override
+  public void addCloudConfig(String clusterName, CloudConfig cloudConfig) {
+    logger.info("Add CloudConfig to cluster {}, CloudConfig is {}.", clusterName,
+        cloudConfig.toString());
+
+    if (!ZKUtil.isClusterSetup(clusterName, _zkClient)) {
+      throw new HelixException("cluster " + clusterName + " is not setup yet");
+    }
+
+    CloudConfig.Builder builder = new CloudConfig.Builder(cloudConfig);
+    CloudConfig cloudConfigBuilder = builder.build();
+
+    ZKHelixDataAccessor accessor =
+        new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_zkClient));
+    Builder keyBuilder = accessor.keyBuilder();
+    accessor.setProperty(keyBuilder.cloudConfig(), cloudConfigBuilder);
+  }
+
+  @Override
+  public void removeCloudConfig(String clusterName) {
+    logger.info("Remove Cloud Config for cluster {}.", clusterName);
+    HelixDataAccessor accessor =
+        new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_zkClient));
+    Builder keyBuilder = accessor.keyBuilder();
+    accessor.removeProperty(keyBuilder.cloudConfig());
+  }
+
   @Override
   public List<String> getStateModelDefs(String clusterName) {
     return _zkClient.getChildren(PropertyPathBuilder.stateModelDef(clusterName));

helix-core/src/main/java/org/apache/helix/model/CloudConfig.java

@@ -30,6 +30,9 @@
  * Cloud configurations
  */
 public class CloudConfig extends HelixProperty {
+
+  public static final String CLOUD_CONFIG_KW = "CloudConfig";
+
   /**
    * Configurable characteristics of a cloud.
    * NOTE: Do NOT use this field name directly, use its corresponding getter/setter in the
@@ -52,39 +55,22 @@ public enum CloudConfigProperty {
 
   /**
    * Instantiate the CloudConfig for the cloud
-   * @param cluster
    */
-  public CloudConfig(String cluster) {
-    super(cluster);
+  private CloudConfig() {
+    super(CLOUD_CONFIG_KW);
   }
 
   /**
-   * Instantiate with a pre-populated record
-   * @param record a ZNRecord corresponding to a cloud configuration
+   * The constructor from the ZNRecord.
+   * @param record
    */
-  public CloudConfig(ZNRecord record) {
-    super(record);
+  private CloudConfig(ZNRecord record) {
+    super(CLOUD_CONFIG_KW);
+    _record.setSimpleFields(record.getSimpleFields());
+    _record.setListFields(record.getListFields());
+    _record.setMapFields(record.getMapFields());
   }
 
-  /**
-   * Instantiate the config using each field individually.
-   * Users should use CloudConfig.Builder to create CloudConfig.
-   * @param cluster
-   * @param enabled
-   * @param cloudID
-   */
-  public CloudConfig(String cluster, boolean enabled, CloudProvider cloudProvider, String cloudID,
-      List<String> cloudInfoSource, String cloudProcessorName) {
-    super(cluster);
-    _record.setBooleanField(CloudConfigProperty.CLOUD_ENABLED.name(), enabled);
-    _record.setSimpleField(CloudConfigProperty.CLOUD_PROVIDER.name(), cloudProvider.name());
-    _record.setSimpleField(CloudConfigProperty.CLOUD_ID.name(), cloudID);
-    if (cloudProvider.equals(CloudProvider.CUSTOMIZED)) {
-      _record
-          .setSimpleField(CloudConfigProperty.CLOUD_INFO_PROCESSOR_NAME.name(), cloudProcessorName);
-      _record.setListField(CloudConfigProperty.CLOUD_INFO_SOURCE.name(), cloudInfoSource);
-    }
-  }
 
   /**
    * Enable/Disable the CLOUD_ENABLED field.
@@ -134,15 +120,6 @@ public List<String> getCloudInfoSources() {
     return _record.getListField(CloudConfigProperty.CLOUD_INFO_SOURCE.name());
   }
 
-  /**
-   * Set the CLOUD_INFO_PROCESSOR_NAME field.
-   * @param cloudInfoProcessorName
-   */
-  public void setCloudInfoFProcessorName(String cloudInfoProcessorName) {
-    _record.setSimpleField(CloudConfigProperty.CLOUD_INFO_PROCESSOR_NAME.name(),
-        cloudInfoProcessorName);
-  }
-
   /**
    * Get the CLOUD_INFO_PROCESSOR_NAME field.
    * @return CLOUD_INFO_PROCESSOR_NAME field.
@@ -151,14 +128,6 @@ public String getCloudInfoProcessorName() {
     return _record.getSimpleField(CloudConfigProperty.CLOUD_INFO_PROCESSOR_NAME.name());
   }
 
-  /**
-   * Set the CLOUD_PROVIDER field.
-   * @param cloudProvider
-   */
-  public void setCloudProvider(CloudProvider cloudProvider) {
-    _record.setSimpleField(CloudConfigProperty.CLOUD_PROVIDER.name(), cloudProvider.name());
-  }
-
   /**
    * Get the CLOUD_PROVIDER field.
    * @return CLOUD_PROVIDER field.
@@ -167,122 +136,104 @@ public String getCloudProvider() {
     return _record.getSimpleField(CloudConfigProperty.CLOUD_PROVIDER.name());
   }
 
+
   public static class Builder {
-    private String _clusterName = null;
-    private CloudProvider _cloudProvider;
-    private boolean _cloudEnabled = DEFAULT_CLOUD_ENABLED;
-    private String _cloudID;
-    private List<String> _cloudInfoSources;
-    private String _cloudInfoProcessorName;
+    private ZNRecord _record;
 
     public CloudConfig build() {
       validate();
-      return new CloudConfig(_clusterName, _cloudEnabled, _cloudProvider, _cloudID,
-          _cloudInfoSources, _cloudInfoProcessorName);
+      return new CloudConfig(_record);
     }
 
     /**
      * Default constructor
      */
     public Builder() {
+      _record = new ZNRecord(CLOUD_CONFIG_KW);
     }
 
     /**
-     * Constructor with Cluster Name as input
-     * @param clusterName
+     * Instantiate with a pre-populated record
+     * @param record a ZNRecord corresponding to a cloud configuration
      */
-    public Builder(String clusterName) {
-      _clusterName = clusterName;
+    public Builder(ZNRecord record) {
+      _record = record;
     }
 
     /**
      * Constructor with CloudConfig as input
      * @param cloudConfig
      */
     public Builder(CloudConfig cloudConfig) {
-      _cloudEnabled = cloudConfig.isCloudEnabled();
-      _cloudProvider = CloudProvider.valueOf(cloudConfig.getCloudProvider());
-      _cloudID = cloudConfig.getCloudID();
-      _cloudInfoSources = cloudConfig.getCloudInfoSources();
-      _cloudInfoProcessorName = cloudConfig.getCloudInfoProcessorName();
-    }
-
-    public Builder setClusterName(String v) {
-      _clusterName = v;
-      return this;
+      _record = cloudConfig.getRecord();
     }
 
     public Builder setCloudEnabled(boolean isEnabled) {
-      _cloudEnabled = isEnabled;
+      _record.setBooleanField(CloudConfigProperty.CLOUD_ENABLED.name(), isEnabled);
       return this;
     }
 
     public Builder setCloudProvider(CloudProvider cloudProvider) {
-      _cloudProvider = cloudProvider;
+      _record.setSimpleField(CloudConfigProperty.CLOUD_PROVIDER.name(), cloudProvider.name());
       return this;
     }
 
-    public Builder setCloudID(String v) {
-      _cloudID = v;
+    public Builder setCloudID(String cloudID) {
+      _record.setSimpleField(CloudConfigProperty.CLOUD_ID.name(), cloudID);
       return this;
     }
 
-    public Builder setCloudInfoSources(List<String> v) {
-      _cloudInfoSources = v;
+    public Builder setCloudInfoSources(List<String> cloudInfoSources) {
+      _record.setListField(CloudConfigProperty.CLOUD_INFO_SOURCE.name(), cloudInfoSources);
       return this;
     }
 
-    public Builder addCloudInfoSource(String v) {
-      if (_cloudInfoSources == null) {
-        _cloudInfoSources = new ArrayList<String>();
+    public Builder addCloudInfoSource(String cloudInfoSource) {
+      if (_record.getListField(CloudConfigProperty.CLOUD_INFO_SOURCE.name()) == null) {
+        _record.setListField(CloudConfigProperty.CLOUD_INFO_SOURCE.name(), new ArrayList<String>());
       }
-      _cloudInfoSources.add(v);
+      List<String> cloudInfoSourcesList = _record.getListField(CloudConfigProperty.CLOUD_INFO_SOURCE.name());
+      cloudInfoSourcesList.add(cloudInfoSource);
+      _record.setListField(CloudConfigProperty.CLOUD_INFO_SOURCE.name(), cloudInfoSourcesList);
       return this;
     }
 
-    public Builder setCloudInfoProcessorName(String v) {
-      _cloudInfoProcessorName = v;
+    public Builder setCloudInfoProcessorName(String cloudInfoProcessorName) {
+      _record.setSimpleField(CloudConfigProperty.CLOUD_INFO_PROCESSOR_NAME.name(),
+          cloudInfoProcessorName);
       return this;
     }
 
-    public String getClusterName() {
-      return _clusterName;
-    }
-
-    public CloudProvider getCloudProvider() {
-      return _cloudProvider;
+    public String getCloudProvider() {
+      return _record.getSimpleField(CloudConfigProperty.CLOUD_PROVIDER.name());
     }
 
     public boolean getCloudEnabled() {
-      return _cloudEnabled;
+      return _record.getBooleanField(CloudConfigProperty.CLOUD_ENABLED.name(),
+          DEFAULT_CLOUD_ENABLED);
     }
 
     public String getCloudID() {
-      return _cloudID;
+      return _record.getSimpleField(CloudConfigProperty.CLOUD_ID.name());
     }
 
     public List<String> getCloudInfoSources() {
-      return _cloudInfoSources;
+      return _record.getListField(CloudConfigProperty.CLOUD_INFO_SOURCE.name());
     }
 
     public String getCloudInfoProcessorName() {
-      return _cloudInfoProcessorName;
+      return _record.getSimpleField(CloudConfigProperty.CLOUD_INFO_PROCESSOR_NAME.name());
     }
 
     private void validate() {
-      if (_cloudEnabled) {
-        if (_cloudID == null) {
-          throw new HelixException(
-              "This Cloud Configuration is Invalid. The CloudID is missing from the config.");
-        }
-        if (_cloudProvider == null) {
+      if (this.getCloudProvider() == null) {
+        throw new HelixException(
+            "This Cloud Configuration is Invalid. The Cloud Provider is missing from the config.");
+      } else if (this.getCloudProvider().equals(CloudProvider.CUSTOMIZED.name())) {
+        if (this.getCloudInfoProcessorName() == null || this.getCloudInfoSources() == null
+            || this.getCloudInfoSources().size() == 0) {
           throw new HelixException(
-              "This Cloud Configuration is Invalid. The Cloud Provider is missing from the config.");
-        } else if (_cloudProvider == CloudProvider.CUSTOMIZED) {
-          if (_cloudInfoProcessorName == null || _cloudInfoSources == null || _cloudInfoSources.size() == 0) {
-            throw new HelixException(
-                "This Cloud Configuration is Invalid. CUSTOMIZED provider has been chosen without defining CloudInfoProcessorName or CloudInfoSources");
-          }
+              "This Cloud Configuration is Invalid. CUSTOMIZED provider has been chosen without defining CloudInfoProcessorName or CloudInfoSources");
         }
       }
     }

helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java

@@ -41,14 +41,17 @@
 import org.apache.helix.HelixConstants;
 import org.apache.helix.HelixException;
 import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.SystemPropertyKeys;
 import org.apache.helix.ZNRecord;
+import org.apache.helix.cloud.constants.CloudProvider;
 import org.apache.helix.manager.zk.ZKHelixAdmin;
 import org.apache.helix.manager.zk.ZKHelixDataAccessor;
 import org.apache.helix.manager.zk.ZNRecordSerializer;
 import org.apache.helix.manager.zk.ZkBaseDataAccessor;
 import org.apache.helix.manager.zk.client.HelixZkClient;
 import org.apache.helix.manager.zk.client.SharedZkClientFactory;
 import org.apache.helix.model.BuiltInStateModelDefinitions;
+import org.apache.helix.model.CloudConfig;
 import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.ClusterConstraints;
 import org.apache.helix.model.ClusterConstraints.ConstraintType;
@@ -176,15 +179,23 @@ public void close() {
     }
   }
 
-  public void addCluster(String clusterName, boolean overwritePrevious) {
+  public void addCluster(String clusterName, boolean overwritePrevious, CloudConfig cloudConfig)
+      throws HelixException {
     _admin.addCluster(clusterName, overwritePrevious);
-
     for (BuiltInStateModelDefinitions def : BuiltInStateModelDefinitions.values()) {
       addStateModelDef(clusterName, def.getStateModelDefinition().getId(),
-                       def.getStateModelDefinition(), overwritePrevious);
+          def.getStateModelDefinition(), overwritePrevious);
+    }
+
+    if (cloudConfig != null) {
+      _admin.addCloudConfig(clusterName, cloudConfig);
     }
   }
 
+  public void addCluster(String clusterName, boolean overwritePrevious) {
+    addCluster(clusterName, overwritePrevious, null);
+  }
+
   public void activateCluster(String clusterName, String grandCluster, boolean enable) {
     if (enable) {
       _admin.addClusterToGrandCluster(clusterName, grandCluster);