From 045deb60fa14220d07cbb4512212465b47969e69 Mon Sep 17 00:00:00 2001 From: Zachary Pinto Date: Wed, 20 Sep 2023 20:29:59 -0700 Subject: [PATCH] Support persist instance info collected by CloudInstanceInformationProcessor during auto-reg (#2622) Add support to persist all instance information collected by CloudInstanceInformationProcessor in CloudInstanceInformation object. Add ability for CloudInstanceInformationProcessor to produce full DOMAIN field instead of appending _instanceName unless last character in CloudInstanceInformation.CloudInstanceField.FAULT_DOMAIN is '='. --- .../api/cloud/CloudInstanceInformationV2.java | 38 +++++++++++++ .../helix/manager/zk/ParticipantManager.java | 39 +++++++++----- .../apache/helix/model/InstanceConfig.java | 54 +++++++++++++++++++ .../apache/helix/util/ConfigStringUtil.java | 2 +- .../CustomCloudInstanceInformation.java | 23 +++++--- ...stomCloudInstanceInformationProcessor.java | 3 +- .../paticipant/TestInstanceAutoJoin.java | 8 ++- .../helix/model/TestInstanceConfig.java | 10 +++- 8 files changed, 152 insertions(+), 25 deletions(-) create mode 100644 helix-core/src/main/java/org/apache/helix/api/cloud/CloudInstanceInformationV2.java diff --git a/helix-core/src/main/java/org/apache/helix/api/cloud/CloudInstanceInformationV2.java b/helix-core/src/main/java/org/apache/helix/api/cloud/CloudInstanceInformationV2.java new file mode 100644 index 0000000000..9d464d1643 --- /dev/null +++ b/helix-core/src/main/java/org/apache/helix/api/cloud/CloudInstanceInformationV2.java @@ -0,0 +1,38 @@ +package org.apache.helix.api.cloud; + +/* + * 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. + */ + +import com.google.common.collect.ImmutableMap; + +/** + * Generic interface for cloud instance information which builds on top of CloudInstanceInformation. + * This interface adds a new method, getAll(), which returns all the key value pairs of a specific cloud instance. + * We call suffix the name of this interface with V2 to preserve backwards compatibility for all classes + * that implement CloudInstanceInformation. + */ +public interface CloudInstanceInformationV2 extends CloudInstanceInformation { + /** + * Get all the key value pairs of a specific cloud instance + * @return A map of all the key value pairs + */ + ImmutableMap getAll(); +} diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java index abc288ec1e..5099ee6acf 100644 --- a/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java +++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ParticipantManager.java @@ -43,6 +43,7 @@ import org.apache.helix.PropertyKey; import org.apache.helix.api.cloud.CloudInstanceInformation; import org.apache.helix.api.cloud.CloudInstanceInformationProcessor; +import org.apache.helix.api.cloud.CloudInstanceInformationV2; import org.apache.helix.messaging.DefaultMessagingService; import org.apache.helix.model.CurrentState; import org.apache.helix.model.HelixConfigScope; @@ -58,6 +59,7 @@ import org.apache.helix.participant.statemachine.StateModelFactory; import org.apache.helix.task.TaskConstants; import org.apache.helix.task.TaskUtil; +import org.apache.helix.util.ConfigStringUtil; import org.apache.helix.zookeeper.api.client.RealmAwareZkClient; import org.apache.helix.zookeeper.datamodel.ZNRecord; import org.apache.helix.zookeeper.datamodel.ZNRecordBucketizer; @@ -208,25 +210,38 @@ private void joinCluster() { InstanceConfig instanceConfig; if (!ZKUtil.isInstanceSetup(_zkclient, _clusterName, _instanceName, _instanceType)) { if (!autoJoin) { - throw new HelixException("Initial cluster structure is not set up for instance: " - + _instanceName + ", instanceType: " + _instanceType); + throw new HelixException( + "Initial cluster structure is not set up for instance: " + _instanceName + + ", instanceType: " + _instanceType); } + + InstanceConfig.Builder instanceConfigBuilder = + _helixManagerProperty.getDefaultInstanceConfigBuilder(); if (!autoRegistration) { LOG.info(_instanceName + " is auto-joining cluster: " + _clusterName); - instanceConfig = - _helixManagerProperty.getDefaultInstanceConfigBuilder().build(_instanceName); + instanceConfig = instanceConfigBuilder.build(_instanceName); } else { LOG.info(_instanceName + " is auto-registering cluster: " + _clusterName); CloudInstanceInformation cloudInstanceInformation = getCloudInstanceInformation(); - String domain = cloudInstanceInformation.get( - CloudInstanceInformation.CloudInstanceField.FAULT_DOMAIN.name()) + _instanceName; - instanceConfig = - _helixManagerProperty.getDefaultInstanceConfigBuilder().build(_instanceName); - instanceConfig.setDomain(domain); + if (cloudInstanceInformation instanceof CloudInstanceInformationV2) { + CloudInstanceInformationV2 cloudInstanceInformationV2 = + (CloudInstanceInformationV2) cloudInstanceInformation; + cloudInstanceInformationV2.getAll().forEach(instanceConfigBuilder::addInstanceInfo); + } + + String cloudInstanceInformationFaultDomain = cloudInstanceInformation.get( + CloudInstanceInformation.CloudInstanceField.FAULT_DOMAIN.name()); + instanceConfig = instanceConfigBuilder.setDomain( + // Previously, the FAULT_DOMAIN was expected to end with the final DOMAIN field key without a value, + // like "rack=25, host=" or "cabinet=A, rack=25, host=". This is because ParticipantManager would append + // the _instanceName to populate the value. This check has been added to preserve backwards compatibility + // while also allowing the auto-registration to construct the full DOMAIN which includes the last value. + cloudInstanceInformationFaultDomain.endsWith(ConfigStringUtil.CONCATENATE_CONFIG_JOINER) + ? cloudInstanceInformationFaultDomain + _instanceName + : cloudInstanceInformationFaultDomain).build(_instanceName); } - instanceConfig - .validateTopologySettingInInstanceConfig(_configAccessor.getClusterConfig(_clusterName), - _instanceName); + instanceConfig.validateTopologySettingInInstanceConfig( + _configAccessor.getClusterConfig(_clusterName), _instanceName); _helixAdmin.addInstance(_clusterName, instanceConfig); } else { _configAccessor.getInstanceConfig(_clusterName, _instanceName) diff --git a/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java b/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java index 252f8254d1..b68250dd37 100644 --- a/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java +++ b/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java @@ -60,6 +60,7 @@ public enum InstanceConfigProperty { DOMAIN, DELAY_REBALANCE_ENABLED, MAX_CONCURRENT_TASK, + INSTANCE_INFO_MAP, INSTANCE_CAPACITY_MAP, TARGET_TASK_THREAD_POOL_SIZE, INSTANCE_OPERATION @@ -607,6 +608,29 @@ public void setTargetTaskThreadPoolSize(int targetTaskThreadPoolSize) targetTaskThreadPoolSize); } + /** + * Get the instance information map from the map fields. + * @return data map if it exists, or empty map + */ + public Map getInstanceInfoMap() { + Map instanceInfoMap = + _record.getMapField(InstanceConfigProperty.INSTANCE_INFO_MAP.name()); + return instanceInfoMap != null ? instanceInfoMap : Collections.emptyMap(); + } + + /** + * Set instanceInfoMap to map of information about the instance that can be used + * to construct the DOMAIN field. + * @param instanceInfoMap Map of information about the instance. ie: { 'rack': 'rack-1', 'host': 'host-1' } + */ + private void setInstanceInfoMap(Map instanceInfoMap) { + if (instanceInfoMap == null) { + _record.getMapFields().remove(InstanceConfigProperty.INSTANCE_INFO_MAP.name()); + } else { + _record.setMapField(InstanceConfigProperty.INSTANCE_INFO_MAP.name(), instanceInfoMap); + } + } + /** * Get the instance capacity information from the map fields. * @return data map if it exists, or empty map @@ -748,6 +772,7 @@ public static class Builder { private int _weight = WEIGHT_NOT_SET; private List _tags = new ArrayList<>(); private boolean _instanceEnabled = HELIX_ENABLED_DEFAULT_VALUE; + private Map _instanceInfoMap; private Map _instanceCapacityMap; /** @@ -794,6 +819,10 @@ public InstanceConfig build(String instanceId) { instanceConfig.setInstanceEnabled(_instanceEnabled); } + if (_instanceInfoMap != null) { + instanceConfig.setInstanceInfoMap(_instanceInfoMap); + } + if (_instanceCapacityMap != null) { instanceConfig.setInstanceCapacityMap(_instanceCapacityMap); } @@ -861,6 +890,31 @@ public Builder setInstanceEnabled(boolean instanceEnabled) { return this; } + /** + * Set the INSTANCE_INFO_MAP for this instance + * @param instanceInfoMap the instance info map + * @return InstanceConfig.Builder + */ + public Builder setInstanceInfoMap(Map instanceInfoMap) { + _instanceInfoMap = instanceInfoMap; + return this; + } + + /** + * Add instance info to the INSTANCE_INFO_MAP. + * Only adds if the key does not already exist. + * @param key the key for the information + * @param value the value the information + * @return InstanceConfig.Builder + */ + public Builder addInstanceInfo(String key, String value) { + if (_instanceInfoMap == null) { + _instanceInfoMap = new HashMap<>(); + } + _instanceInfoMap.putIfAbsent(key, value); + return this; + } + /** * Set the capacity map for this instance * @param instanceCapacityMap the capacity map diff --git a/helix-core/src/main/java/org/apache/helix/util/ConfigStringUtil.java b/helix-core/src/main/java/org/apache/helix/util/ConfigStringUtil.java index 68f42026c5..e9d0aa4510 100644 --- a/helix-core/src/main/java/org/apache/helix/util/ConfigStringUtil.java +++ b/helix-core/src/main/java/org/apache/helix/util/ConfigStringUtil.java @@ -25,7 +25,7 @@ public final class ConfigStringUtil { private static final String CONCATENATE_CONFIG_SPLITTER = ","; - private static final String CONCATENATE_CONFIG_JOINER = "="; + public static final String CONCATENATE_CONFIG_JOINER = "="; private ConfigStringUtil() { throw new java.lang.UnsupportedOperationException( diff --git a/helix-core/src/test/java/org/apache/helix/integration/paticipant/CustomCloudInstanceInformation.java b/helix-core/src/test/java/org/apache/helix/integration/paticipant/CustomCloudInstanceInformation.java index 02f6bc2c0e..bfbf0866a6 100644 --- a/helix-core/src/test/java/org/apache/helix/integration/paticipant/CustomCloudInstanceInformation.java +++ b/helix-core/src/test/java/org/apache/helix/integration/paticipant/CustomCloudInstanceInformation.java @@ -19,24 +19,31 @@ * under the License. */ +import com.google.common.collect.ImmutableMap; import org.apache.helix.api.cloud.CloudInstanceInformation; +import org.apache.helix.api.cloud.CloudInstanceInformationV2; /** * This is a custom implementation of CloudInstanceInformation. It is used to test the functionality * of Helix node auto-registration. */ -public class CustomCloudInstanceInformation implements CloudInstanceInformation { - private final String _faultDomain; +public class CustomCloudInstanceInformation implements CloudInstanceInformationV2 { - public CustomCloudInstanceInformation(String faultDomain) { - _faultDomain = faultDomain; + public static final ImmutableMap _cloudInstanceInfo = + ImmutableMap.of(CloudInstanceInformation.CloudInstanceField.FAULT_DOMAIN.name(), + "mz=0, host=localhost, container=containerId", "MAINTENANCE_ZONE", "0", "INSTANCE_NAME", + "localhost_something"); + + public CustomCloudInstanceInformation() { } @Override public String get(String key) { - if (key.equals(CloudInstanceField.FAULT_DOMAIN.name())) { - return _faultDomain; - } - return null; + return _cloudInstanceInfo.get(key); + } + + @Override + public ImmutableMap getAll() { + return _cloudInstanceInfo; } } diff --git a/helix-core/src/test/java/org/apache/helix/integration/paticipant/CustomCloudInstanceInformationProcessor.java b/helix-core/src/test/java/org/apache/helix/integration/paticipant/CustomCloudInstanceInformationProcessor.java index 2db41db2ee..37e8e40171 100644 --- a/helix-core/src/test/java/org/apache/helix/integration/paticipant/CustomCloudInstanceInformationProcessor.java +++ b/helix-core/src/test/java/org/apache/helix/integration/paticipant/CustomCloudInstanceInformationProcessor.java @@ -31,6 +31,7 @@ * It is used to test the functionality of Helix node auto-registration. */ public class CustomCloudInstanceInformationProcessor implements CloudInstanceInformationProcessor { + public CustomCloudInstanceInformationProcessor(HelixCloudProperty helixCloudProperty) { } @@ -41,6 +42,6 @@ public List fetchCloudInstanceInformation() { @Override public CloudInstanceInformation parseCloudInstanceInformation(List responses) { - return new CustomCloudInstanceInformation("rack=A:123, host="); + return new CustomCloudInstanceInformation(); } } \ No newline at end of file diff --git a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestInstanceAutoJoin.java b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestInstanceAutoJoin.java index f12cec6ba8..8885549479 100644 --- a/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestInstanceAutoJoin.java +++ b/helix-core/src/test/java/org/apache/helix/integration/paticipant/TestInstanceAutoJoin.java @@ -8,6 +8,7 @@ import org.apache.helix.HelixManagerProperty; import org.apache.helix.PropertyKey; import org.apache.helix.TestHelper; +import org.apache.helix.api.cloud.CloudInstanceInformation; import org.apache.helix.cloud.constants.CloudProvider; import org.apache.helix.controller.rebalancer.strategy.CrushEdRebalanceStrategy; import org.apache.helix.integration.common.ZkStandAloneCMTestBase; @@ -211,8 +212,11 @@ public void testAutoRegistrationCustomizedFullyQualifiedInfoProcessorPath() thro // Check that live instance is added and instance config is populated with correct domain. return null != manager.getHelixDataAccessor() .getProperty(accessor.keyBuilder().liveInstance(instance5)) && manager.getConfigAccessor() - .getInstanceConfig(CLUSTER_NAME, instance5).getDomainAsString() - .equals("rack=A:123, host=" + instance5); + .getInstanceConfig(CLUSTER_NAME, instance5).getDomainAsString().equals( + CustomCloudInstanceInformation._cloudInstanceInfo.get( + CloudInstanceInformation.CloudInstanceField.FAULT_DOMAIN.name())) + && manager.getConfigAccessor().getInstanceConfig(CLUSTER_NAME, instance5) + .getInstanceInfoMap().equals(CustomCloudInstanceInformation._cloudInstanceInfo); }, 2000)); autoParticipant.syncStop(); diff --git a/helix-core/src/test/java/org/apache/helix/model/TestInstanceConfig.java b/helix-core/src/test/java/org/apache/helix/model/TestInstanceConfig.java index 2fee5ae3cd..b8e6569f5d 100644 --- a/helix-core/src/test/java/org/apache/helix/model/TestInstanceConfig.java +++ b/helix-core/src/test/java/org/apache/helix/model/TestInstanceConfig.java @@ -20,6 +20,7 @@ */ import java.util.Collections; +import java.util.HashMap; import java.util.Map; import com.google.common.collect.ImmutableMap; @@ -173,11 +174,16 @@ public void testSetTargetTaskThreadPoolSizeIllegalArgument() { @Test public void testInstanceConfigBuilder() { + + Map instanceInfoMap = new HashMap<>(); + instanceInfoMap.put("CAGE", "H"); Map capacityDataMap = ImmutableMap.of("weight1", 1); InstanceConfig instanceConfig = new InstanceConfig.Builder().setHostName("testHost").setPort("1234").setDomain("foo=bar") .setWeight(100).setInstanceEnabled(true).addTag("tag1").addTag("tag2") - .setInstanceEnabled(false).setInstanceCapacityMap(capacityDataMap).build("instance1"); + .setInstanceEnabled(false).setInstanceInfoMap(instanceInfoMap) + .addInstanceInfo("CAGE", "G").addInstanceInfo("CABINET", "30") + .setInstanceCapacityMap(capacityDataMap).build("instance1"); Assert.assertEquals(instanceConfig.getId(), "instance1"); Assert.assertEquals(instanceConfig.getHostName(), "testHost"); @@ -187,6 +193,8 @@ public void testInstanceConfigBuilder() { Assert.assertTrue(instanceConfig.getTags().contains("tag1")); Assert.assertTrue(instanceConfig.getTags().contains("tag2")); Assert.assertFalse(instanceConfig.getInstanceEnabled()); + Assert.assertEquals(instanceConfig.getInstanceInfoMap().get("CAGE"), "H"); + Assert.assertEquals(instanceConfig.getInstanceInfoMap().get("CABINET"), "30"); Assert.assertEquals(instanceConfig.getInstanceCapacityMap().get("weight1"), Integer.valueOf(1)); } }