Add Etcd membership module

conf/etcd/etcd.conf.template

@@ -0,0 +1,69 @@
+# This is the configuration file to start a etcd instance
+# e.g. /usr/local/bin/etcd --config-file /etc/etcd/etcd.conf 
+# *******README******
+# To make etcd a linux service:
+# After installation of etcd, make sure etcd and etcdctl
+# are available in /usr/local/bin
+# To make etcd a linux service:
+# Copy alluxio/conf/etcd/etcd.service.template to /etc/systemd/system/etcd.service
+# Copy alluxio/conf/etcd/etcd.conf.template to /etc/etcd/etcd.conf
+# For each etcd instance, change the config params in etcd.conf
+# accordingly.
+# And do:
+# #systemctl daemon-reload 
+# Then etcd could be registered as a linux service
+# e.g.
+# Check status
+# #service etcd status
+# Start etcd
+# #service etcd start
+# Stop etcd
+# #service etcd stop
+
+
+# Human-readable name for this member.
+#name: 'etcd1'
+
+# Path to the data directory.
+data-dir: /etcd-data-dir/data
+
+# Path to the dedicated wal directory.
+wal-dir: /etcd-data-dir/wal
+
+
+# List of comma separated URLs to listen on for peer traffic.
+#give ip/hostname of this etcd instance
+listen-peer-urls: http://<hostname_or_ip>:2380
+
+# List of comma separated URLs to listen on for client traffic.
+#give ip/hostname of this etcd instance
+listen-client-urls: http://<hostname_or_ip>:2379,http://127.0.0.1:2379
+
+# List of this member's peer URLs to advertise to the rest of the cluster.
+# The URLs needed to be a comma-separated list.
+#give ip/hostname of this etcd instance for remote etcd members communication
+initial-advertise-peer-urls: http://<hostname_or_ip>:2380
+
+# List of this member's client URLs to advertise to the public.
+# The URLs needed to be a comma-separated list.
+#give ip/hostname of this etcd instance for etcd client communication
+advertise-client-urls: http://<hostname_or_ip>:2379
+
+# Initial cluster configuration for bootstrapping.
+#give all ip/hostnames of members of initial etcd cluster
+initial-cluster: etcd0=http://<hostname_or_ip>:2380,etcd1=http://<hostname_or_ip>:2380,etcd2=http://<hostname_or_ip>:2380
+
+# Initial cluster token for the etcd cluster during bootstrap.
+#initial-cluster-token: 'etcd-cluster-1'
+
+# Initial cluster state ('new' or 'existing').
+initial-cluster-state: 'new'
+
+# Enable debug-level logging for etcd.
+#log-level: debug
+
+#logger: zap
+
+# Specify 'stdout' or 'stderr' to skip journald logging even when running under systemd.
+# log-outputs: [stderr]
+

conf/etcd/etcd.service.template

@@ -0,0 +1,11 @@
+[Unit]
+Description=Etcd Service
+
+[Service]	
+ExecStart=/usr/local/bin/etcd --config-file /etc/etcd/etcd.conf
+KillSignal=SIGTERM
+StandardOutput=append:/var/log/etcd.log
+StandardError=append:/var/log/etcd.err
+
+[Install]
+WantedBy=default.target

dora/core/client/fs/src/main/java/alluxio/client/file/FileSystemContext.java

@@ -34,6 +34,8 @@
 import alluxio.grpc.GrpcServerAddress;
 import alluxio.master.MasterClientContext;
 import alluxio.master.MasterInquireClient;
+import alluxio.membership.MembershipManager;
+import alluxio.membership.NoOpMembershipManager;
 import alluxio.metrics.MetricsSystem;
 import alluxio.network.netty.NettyChannelPool;
 import alluxio.network.netty.NettyClient;
@@ -154,6 +156,8 @@ public class FileSystemContext implements Closeable {
    */
   private volatile ConcurrentHashMap<ClientPoolKey, BlockWorkerClientPool>
       mBlockWorkerClientPoolMap;
+  @Nullable
+  private MembershipManager mMembershipManager;
 
   /**
    * Indicates whether the {@link #mLocalWorker} field has been lazily initialized yet.
@@ -443,6 +447,11 @@ protected synchronized void initContext(ClientContext ctx,
     mBlockMasterClientPool = new BlockMasterClientPool(mMasterClientContext);
     mBlockWorkerClientPoolMap = new ConcurrentHashMap<>();
     mUriValidationEnabled = ctx.getUriValidationEnabled();
+    try {
+      mMembershipManager = MembershipManager.Factory.create(getClusterConf());
+    } catch (IOException ex) {
+      LOG.error("Failed to set membership manager.", ex);
+    }
   }
 
   /**
@@ -490,6 +499,12 @@ private synchronized void closeContext() throws IOException {
       if (mMetricsEnabled) {
         MetricsHeartbeatContext.removeHeartbeat(getClientContext());
       }
+      LOG.debug("Closing membership manager.");
+      try (AutoCloseable ignoredCloser = mMembershipManager) {
+        // do nothing as we are closing
+      } catch (Exception e) {
+        throw new IOException(e);
+      }
     } else {
       LOG.warn("Attempted to close FileSystemContext which has already been closed or not "
           + "initialized.");
@@ -864,6 +879,17 @@ public List<BlockWorkerInfo> getCachedWorkers() throws IOException {
    * @return the info of all block workers
    */
   protected List<BlockWorkerInfo> getAllWorkers() throws IOException {
+    // TODO(lucy) once ConfigHashSync reinit is gotten rid of, will remove the blockReinit
+    // guard altogether
+    try (ReinitBlockerResource r = blockReinit()) {
+      // Use membership mgr
+      if (mMembershipManager != null && !(mMembershipManager instanceof NoOpMembershipManager)) {
+        return mMembershipManager.getAllMembers().stream()
+            .map(w -> new BlockWorkerInfo(w.getAddress(), w.getCapacityBytes(), w.getUsedBytes()))
+            .collect(toList());
+      }
+    }
+    // Fall back to old way
     try (CloseableResource<BlockMasterClient> masterClientResource =
              acquireBlockMasterClientResource()) {
       return masterClientResource.get().getWorkerInfoList().stream()

dora/core/common/pom.xml

@@ -87,6 +87,10 @@
       <groupId>io.dropwizard.metrics</groupId>
       <artifactId>metrics-jvm</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.etcd</groupId>
+      <artifactId>jetcd-core</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-core</artifactId>

dora/core/common/src/main/java/alluxio/conf/PropertyKey.java

@@ -48,6 +48,7 @@
 import alluxio.master.metastore.MetastoreType;
 import alluxio.master.metastore.rocks.DataBlockIndexType;
 import alluxio.master.metastore.rocks.IndexType;
+import alluxio.membership.MembershipType;
 import alluxio.network.ChannelType;
 import alluxio.network.netty.FileTransferType;
 import alluxio.security.authentication.AuthType;
@@ -5505,6 +5506,25 @@ public String toString() {
           .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
           .setScope(Scope.WORKER)
           .build();
+  public static final PropertyKey WORKER_MEMBERSHIP_MANAGER_TYPE =
+      enumBuilder(Name.WORKER_MEMBERSHIP_MANAGER_TYPE, MembershipType.class)
+          .setDefaultValue(MembershipType.NOOP.name())
+          .setDescription("Type of membership manager used for workers."
+              + "Choose STATIC for pre-configured members."
+              + "Choose ETCD for using etcd for membership management"
+              + "Default is NOOP which does not enable membership manager at all")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.ENFORCE)
+          .setScope(Scope.ALL)
+          .build();
+  public static final PropertyKey WORKER_STATIC_MEMBERSHIP_MANAGER_CONFIG_FILE =
+      stringBuilder(Name.WORKER_STATIC_MEMBERSHIP_MANAGER_CONFIG_FILE)
+          .setDefaultValue(format("${%s}/workers", Name.CONF_DIR))
+          .setDescription("Absolute path of the config file for list"
+              + "of worker hostnames/IPs for the cluster. "
+              + Name.WORKER_MEMBERSHIP_MANAGER_TYPE + " needs to be set"
+              + " to STATIC first.")
+          .setScope(Scope.ALL)
+          .build();
 
   //
   // Proxy related properties
@@ -7626,6 +7646,19 @@ public String toString() {
       stringBuilder(Name.ZOOKEEPER_JOB_LEADER_PATH)
           .setDefaultValue("/alluxio/job_leader").build();
 
+  //
+  // Membership related properties
+  //
+  public static final PropertyKey ALLUXIO_CLUSTER_NAME =
+      stringBuilder(Name.ALLUXIO_CLUSTER_NAME)
+          .setDefaultValue("DefaultAlluxioCluster").build();
+  public static final PropertyKey ETCD_ENDPOINTS =
+      listBuilder(Name.ETCD_ENDPOINTS)
+          .setDescription("A list of comma-separated http://host:port addresses of "
+                  + "etcd cluster (e.g. http://localhost:2379,http://etcd1:2379)")
+          .setScope(Scope.ALL)
+          .build();
+
   //
   // JVM Monitor related properties
   //
@@ -8993,6 +9026,10 @@ public static final class Name {
     public static final String WORKER_UFS_INSTREAM_CACHE_MAX_SIZE =
         "alluxio.worker.ufs.instream.cache.max.size";
     public static final String WORKER_WHITELIST = "alluxio.worker.whitelist";
+    public static final String WORKER_MEMBERSHIP_MANAGER_TYPE =
+        "alluxio.worker.membership.manager.type";
+    public static final String WORKER_STATIC_MEMBERSHIP_MANAGER_CONFIG_FILE =
+        "alluxio.worker.static.membership.manager.config.file";
 
     //
     // Proxy related properties
@@ -9480,6 +9517,10 @@ public static final class Name {
     public static final String ZOOKEEPER_JOB_ELECTION_PATH = "alluxio.zookeeper.job.election.path";
     public static final String ZOOKEEPER_JOB_LEADER_PATH = "alluxio.zookeeper.job.leader.path";
 
+    // Membership related properties
+    public static final String ALLUXIO_CLUSTER_NAME = "alluxio.cluster.name";
+    public static final String ETCD_ENDPOINTS = "alluxio.etcd.endpoints";
+
     //
     // JVM Monitor related properties
     //