accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [2/3] git commit: ACCUMULO-2582 Initial implementation of insight into replication
Date Fri, 23 May 2014 02:12:47 GMT
ACCUMULO-2582 Initial implementation of insight into replication


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/9d8d347c
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/9d8d347c
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/9d8d347c

Branch: refs/heads/ACCUMULO-378
Commit: 9d8d347c88fb99fb21382a9bdce2658c009ea9bc
Parents: db3818b
Author: Josh Elser <elserj@apache.org>
Authored: Thu May 22 21:47:03 2014 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Thu May 22 22:03:21 2014 -0400

----------------------------------------------------------------------
 server/monitor/pom.xml                          |  10 +-
 .../org/apache/accumulo/monitor/Monitor.java    |   2 +
 .../accumulo/monitor/servlets/BasicServlet.java |   1 +
 .../monitor/servlets/ReplicationServlet.java    | 191 +++++++++++++++++++
 4 files changed, 198 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d8d347c/server/monitor/pom.xml
----------------------------------------------------------------------
diff --git a/server/monitor/pom.xml b/server/monitor/pom.xml
index 411812c..0ff4f19 100644
--- a/server/monitor/pom.xml
+++ b/server/monitor/pom.xml
@@ -88,18 +88,16 @@
       <artifactId>jetty</artifactId>
     </dependency>
     <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
-      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-log4j12</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
   </dependencies>

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d8d347c/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
index e6617d0..268516c 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
@@ -58,6 +58,7 @@ import org.apache.accumulo.monitor.servlets.LogServlet;
 import org.apache.accumulo.monitor.servlets.MasterServlet;
 import org.apache.accumulo.monitor.servlets.OperationServlet;
 import org.apache.accumulo.monitor.servlets.ProblemServlet;
+import org.apache.accumulo.monitor.servlets.ReplicationServlet;
 import org.apache.accumulo.monitor.servlets.ShellServlet;
 import org.apache.accumulo.monitor.servlets.TServersServlet;
 import org.apache.accumulo.monitor.servlets.TablesServlet;
@@ -440,6 +441,7 @@ public class Monitor {
     server.addServlet(Summary.class, "/trace/summary");
     server.addServlet(ListType.class, "/trace/listType");
     server.addServlet(ShowTrace.class, "/trace/show");
+    server.addServlet(ReplicationServlet.class, "/replication");
     if (server.isUsingSsl())
       server.addServlet(ShellServlet.class, "/shell");
     server.start();

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d8d347c/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
index 86a84b9..1abb181 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
@@ -189,6 +189,7 @@ abstract public class BasicServlet extends HttpServlet {
     sb.append("<a href='/gc'>Garbage&nbsp;Collector</a><br />\n");
     sb.append("<a href='/tables'>Tables</a><br />\n");
     sb.append("<a href='/trace/summary?minutes=10'>Recent&nbsp;Traces</a><br
/>\n");
+    sb.append("<a href='/replication'>Replication</a><br />\n");
     List<DedupedLogEvent> dedupedLogEvents = LogService.getInstance().getEvents();
     int numLogs = dedupedLogEvents.size();
     boolean logsHaveError = false;

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d8d347c/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
new file mode 100644
index 0000000..3fda43f
--- /dev/null
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
@@ -0,0 +1,191 @@
+/*
+ * 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.accumulo.monitor.servlets;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.client.replication.ReplicaSystem;
+import org.apache.accumulo.core.client.replication.ReplicaSystemFactory;
+import org.apache.accumulo.core.client.replication.ReplicationTable;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTarget;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.Credentials;
+import org.apache.accumulo.monitor.util.Table;
+import org.apache.accumulo.monitor.util.celltypes.NumberType;
+import org.apache.accumulo.server.client.HdfsZooInstance;
+import org.apache.accumulo.server.security.SystemCredentials;
+import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Maps;
+
+/**
+ * 
+ */
+public class ReplicationServlet extends BasicServlet {
+  private static final Logger log = LoggerFactory.getLogger(ReplicationServlet.class);
+
+  private static final long serialVersionUID = 1L;
+
+  @Override
+  protected String getTitle(HttpServletRequest req) {
+    return "Replication Overview";
+  }
+  
+  @Override
+  protected void pageBody(HttpServletRequest req, HttpServletResponse response, StringBuilder
sb) throws Exception {
+    Instance inst = HdfsZooInstance.getInstance();
+    Credentials creds = SystemCredentials.get();
+    Connector conn = inst.getConnector(creds.getPrincipal(), creds.getToken());
+
+    TableOperations tops = conn.tableOperations();
+    if (!tops.exists(ReplicationTable.NAME)) {
+      banner(sb, "", "Replication table does not yet exist");
+      return;
+    }
+
+    Table replicationStats = new Table("replicationStats", "Replication Status");
+    replicationStats.addSortableColumn("Table");
+    replicationStats.addSortableColumn("Peer");
+    replicationStats.addSortableColumn("Remote Identifier");
+    replicationStats.addSortableColumn("ReplicaSystem Type");
+    replicationStats.addSortableColumn("Files needing replication", new NumberType<Long>(),
null);
+
+    Map<String,String> properties = conn.instanceOperations().getSystemConfiguration();
+    Map<String,String> peers = new HashMap<>();
+    String definedPeersPrefix = Property.REPLICATION_PEERS.getKey();
+
+    // Get the defined peers and what ReplicaSystem impl they're using
+    for (Entry<String,String> property : properties.entrySet()) {
+      if (property.getKey().startsWith(definedPeersPrefix)) {
+        String peerName = property.getKey().substring(definedPeersPrefix.length());
+        ReplicaSystem replica;
+        try {
+         replica = ReplicaSystemFactory.get(property.getValue());
+        } catch (Exception e) {
+          log.warn("Could not instantiate ReplicaSystem for {} with configuration {}", property.getKey(),
property.getValue(), e);
+          continue;
+        }
+
+        peers.put(peerName, replica.getClass().getName());
+      }
+    }
+
+    final String targetPrefix = Property.TABLE_REPLICATION_TARGETS.getKey();
+
+    // The total set of configured targets
+    Set<ReplicationTarget> allConfiguredTargets = new HashSet<>();
+
+    // Number of files per target we have to replicate
+    Map<ReplicationTarget,Long> targetCounts = new HashMap<>();
+
+    Map<String,String> tableNameToId = tops.tableIdMap();
+    Map<String,String> tableIdToName = invert(tableNameToId);
+
+    for (String table : tops.list()) {
+      if (MetadataTable.NAME.equals(table) || RootTable.NAME.equals(table)) {
+        continue;
+      }
+      String localId = tableNameToId.get(table);
+      if (null == localId) {
+        log.trace("Could not determine ID for {}", table);
+        continue;
+      }
+
+      Iterable<Entry<String,String>> propertiesForTable = tops.getProperties(table);
+      for (Entry<String,String> prop : propertiesForTable) {
+        if (prop.getKey().startsWith(targetPrefix)) {
+          String peerName = prop.getKey().substring(targetPrefix.length());
+          String remoteIdentifier = prop.getValue();
+          ReplicationTarget target = new ReplicationTarget(peerName, remoteIdentifier, localId);
+
+          allConfiguredTargets.add(target);
+        }
+      }
+    }
+
+    // Read over the queued work
+    BatchScanner bs = conn.createBatchScanner(ReplicationTable.NAME, Authorizations.EMPTY,
4);
+    bs.setRanges(Collections.singleton(new Range()));
+    WorkSection.limit(bs);
+    try {
+      Text buffer = new Text();
+      for (Entry<Key,Value> entry : bs) {
+        Key k = entry.getKey();
+        k.getColumnQualifier(buffer);
+        ReplicationTarget target = ReplicationTarget.from(buffer);
+
+        Long count = targetCounts.get(target);
+        if (null == count) {
+          targetCounts.put(target, Long.valueOf(1l));
+        } else {
+          targetCounts.put(target, count + 1);
+        }
+      }
+    } finally {
+      bs.close();
+    }
+
+    for (ReplicationTarget configuredTarget : allConfiguredTargets) {
+      String tableName = tableIdToName.get(configuredTarget.getSourceTableId());
+      if (null == tableName) {
+        log.trace("Could not determine table name from id {}", configuredTarget.getSourceTableId());
+        continue;
+      }
+
+      String replicaSystemClass = peers.get(configuredTarget.getPeerName());
+      if (null == replicaSystemClass) {
+        log.trace("Could not determine configured ReplicaSystem for {}", configuredTarget.getPeerName());
+        continue;
+      }
+
+      Long numFiles = targetCounts.get(configuredTarget);
+
+      replicationStats.addRow(tableName, configuredTarget.getPeerName(), configuredTarget.getRemoteIdentifier(),
replicaSystemClass, (null == numFiles) ? 0 : numFiles); 
+    }
+
+    replicationStats.generate(req, sb);
+  }
+
+  protected Map<String,String> invert(Map<String,String> map) {
+    Map<String,String> newMap = Maps.newHashMapWithExpectedSize(map.size());
+    for(Entry<String,String> entry : map.entrySet()) {
+      newMap.put(entry.getValue(), entry.getKey());
+    }
+    return newMap;
+  }
+}


Mime
View raw message