accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [08/10] accumulo git commit: ACCUMULO-3320 Fetch the necessary info from the master and tservers in the GC
Date Mon, 10 Nov 2014 21:46:27 GMT
ACCUMULO-3320 Fetch the necessary info from the master and tservers in the GC


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

Branch: refs/heads/master
Commit: 2b3473469d38241f3a9ada79d968853801d91597
Parents: a754e1b
Author: Josh Elser <elserj@apache.org>
Authored: Sat Nov 8 22:03:21 2014 -0500
Committer: Josh Elser <elserj@apache.org>
Committed: Mon Nov 10 13:34:51 2014 -0800

----------------------------------------------------------------------
 .../gc/GarbageCollectWriteAheadLogs.java        | 129 ++++++++++++++++---
 .../gc/GarbageCollectWriteAheadLogsTest.java    |  79 ++++++++++++
 .../master/MasterClientServiceHandler.java      |  13 +-
 .../apache/accumulo/tserver/TabletServer.java   |   7 +
 .../test/performance/thrift/NullTserver.java    |   5 +
 5 files changed, 215 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/2b347346/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
index e96fee3..50256b2 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
@@ -36,11 +36,13 @@ import org.apache.accumulo.core.client.Instance;
 import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
+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.gc.thrift.GCStatus;
 import org.apache.accumulo.core.gc.thrift.GcCycleStats;
+import org.apache.accumulo.core.master.thrift.MasterClientService;
 import org.apache.accumulo.core.metadata.MetadataTable;
 import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
 import org.apache.accumulo.core.protobuf.ProtobufUtil;
@@ -51,12 +53,14 @@ import org.apache.accumulo.core.replication.StatusUtil;
 import org.apache.accumulo.core.replication.proto.Replication.Status;
 import org.apache.accumulo.core.security.Authorizations;
 import org.apache.accumulo.core.security.Credentials;
+import org.apache.accumulo.core.security.thrift.TCredentials;
 import org.apache.accumulo.core.tabletserver.log.LogEntry;
 import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
 import org.apache.accumulo.core.tabletserver.thrift.TabletClientService.Client;
 import org.apache.accumulo.core.trace.Span;
 import org.apache.accumulo.core.trace.Trace;
 import org.apache.accumulo.core.trace.Tracer;
+import org.apache.accumulo.core.trace.thrift.TInfo;
 import org.apache.accumulo.core.util.AddressUtil;
 import org.apache.accumulo.core.util.ThriftUtil;
 import org.apache.accumulo.core.zookeeper.ZooUtil;
@@ -69,6 +73,7 @@ import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.thrift.TException;
+import org.apache.thrift.transport.TTransportException;
 import org.apache.zookeeper.KeeperException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -355,12 +360,23 @@ public class GarbageCollectWriteAheadLogs {
       throw new IllegalArgumentException(e);
     }
 
+
+    final AccumuloConfiguration conf = new ServerConfigurationFactory(instance).getConfiguration();
+    final TInfo tinfo = Tracer.traceInfo();
+    final TCredentials tcreds = SystemCredentials.get().toThrift(getInstance());
+    Set<String> activeWals = getActiveWals(conf, tinfo, tcreds);
+
     int count = 0;
 
     Iterator<Entry<String,Path>> walIter = nameToFileMap.entrySet().iterator();
     while (walIter.hasNext()) {
       Entry<String,Path> wal = walIter.next();
       String fullPath = wal.getValue().toString();
+      if (null == activeWals) {
+        log.debug("Could not contact servers to determine if WAL is still in use");
+        walIter.remove();
+        sortedWALogs.remove(wal.getKey());
+      }
       if (neededByReplication(conn, fullPath)) {
         log.debug("Removing WAL from candidate deletion as it is still needed for replication:
{} ", fullPath);
         // If we haven't already removed it, check to see if this WAL is
@@ -378,6 +394,102 @@ public class GarbageCollectWriteAheadLogs {
     return count;
   }
 
+  private String getMasterAddress() {
+    try {
+      List<String> locations = getInstance().getMasterLocations();
+      if (locations.size() == 0)
+        return null;
+      return locations.get(0);
+    } catch (Exception e) {
+      log.warn("Failed to obtain master host " + e);
+    }
+
+    return null;
+  }
+
+  private MasterClientService.Client getMasterConnection(AccumuloConfiguration conf) {
+    final String address = getMasterAddress();
+    try {
+      if (address == null) {
+        log.warn("Could not fetch Master address");
+        return null;
+      }
+      return ThriftUtil.getClient(new MasterClientService.Client.Factory(), address, Property.GENERAL_RPC_TIMEOUT,
conf);
+    } catch (Exception e) {
+      log.warn("Issue with masterConnection (" + address + ") " + e, e);
+    }
+    return null;
+  }
+
+  /**
+   * Fetch the set of WALs in use by tabletservers
+   *
+   * @return Set of WALs in use by tservers, null if they cannot be computed for some reason
+   */
+  protected Set<String> getActiveWals(AccumuloConfiguration conf, TInfo tinfo, TCredentials
tcreds) {
+    List<String> tservers = getActiveTservers(conf, tinfo, tcreds);
+
+    // Compute the total set of WALs used by tservers
+    Set<String> walogs = null;
+    if (null != tservers) {
+      walogs = new HashSet<String>();
+      for (String tserver : tservers) {
+        HostAndPort address = HostAndPort.fromString(tserver);
+        List<String> activeWalsForServer = getActiveWalsForServer(conf, tinfo, tcreds,
address);
+        if (null == activeWalsForServer) {
+          log.debug("Could not fetch active wals from " + address);
+          return null;
+        }
+        log.debug("Got active wals for " + address + ", " + activeWalsForServer);
+        walogs.addAll(activeWalsForServer);
+      }
+    }
+
+    return walogs;
+  }
+
+  /**
+   * Get the active tabletservers as seen by the master.
+   *
+   * @return The active tabletservers, null if they can't be computed.
+   */
+  protected List<String> getActiveTservers(AccumuloConfiguration conf, TInfo tinfo,
TCredentials tcreds) {
+    MasterClientService.Client client = null;
+
+    List<String> tservers = null;
+    try {
+      client = getMasterConnection(conf);
+
+      if (null != client) {
+        tservers = client.getActiveTservers(tinfo, tcreds);
+      }
+    } catch (TException e) {
+      // If we can't fetch the tabletservers, we can't fetch any active WALs
+      log.warn("Failed to fetch active tabletservers from the master", e);
+      return null;
+    } finally {
+      ThriftUtil.returnClient(client);
+    }
+
+    return tservers;
+  }
+
+  protected List<String> getActiveWalsForServer(AccumuloConfiguration conf, TInfo tinfo,
TCredentials tcreds, HostAndPort server) {
+    TabletClientService.Client tserverClient = null;
+    try {
+      tserverClient = ThriftUtil.getClient(new TabletClientService.Client.Factory(), server,
conf);
+      return tserverClient.getActiveLogs(tinfo, tcreds);
+    } catch (TTransportException e) {
+      log.warn("Failed to fetch active write-ahead logs from " + server, e);
+      return null;
+    } catch (TException e) {
+      log.warn("Failed to fetch active write-ahead logs from " + server, e);
+      return null;
+    } finally {
+      ThriftUtil.returnClient(tserverClient);
+    }
+  }
+
   /**
    * Determine if the given WAL is needed for replication
    *
@@ -388,23 +500,6 @@ public class GarbageCollectWriteAheadLogs {
   protected boolean neededByReplication(Connector conn, String wal) {
     log.info("Checking replication table for " + wal);
 
-    // try {
-    // log.info("Current state of Metadata table");
-    // for (Entry<Key,Value> entry : conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY))
{
-    // log.info(entry.getKey().toStringNoTruncate() + "=" + TextFormat.shortDebugString(Status.parseFrom(entry.getValue().get())));
-    // }
-    // } catch (Exception e) {
-    // log.error("Could not read metadata table");
-    // }
-    // try {
-    // log.info("Current state of replication table");
-    // for (Entry<Key,Value> entry : conn.createScanner(ReplicationTable.NAME, Authorizations.EMPTY))
{
-    // log.info(entry.getKey().toStringNoTruncate() + "=" + TextFormat.shortDebugString(Status.parseFrom(entry.getValue().get())));
-    // }
-    // } catch (Exception e) {
-    // log.error("Could not read replication table");
-    // }
-
     Iterable<Entry<Key,Value>> iter = getReplicationStatusForFile(conn, wal);
 
     // TODO Push down this filter to the tserver to only return records

http://git-wip-us.apache.org/repos/asf/accumulo/blob/2b347346/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
----------------------------------------------------------------------
diff --git a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
index 71e5f7d..780907f 100644
--- a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
+++ b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
@@ -27,12 +27,15 @@ import static org.junit.Assert.assertTrue;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 import java.util.UUID;
 
 import org.apache.accumulo.core.client.BatchWriter;
@@ -41,6 +44,7 @@ import org.apache.accumulo.core.client.Connector;
 import org.apache.accumulo.core.client.Instance;
 import org.apache.accumulo.core.client.mock.MockInstance;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.data.Value;
@@ -54,10 +58,13 @@ import org.apache.accumulo.core.replication.ReplicationTable;
 import org.apache.accumulo.core.replication.StatusUtil;
 import org.apache.accumulo.core.replication.proto.Replication.Status;
 import org.apache.accumulo.core.security.Credentials;
+import org.apache.accumulo.core.security.thrift.TCredentials;
+import org.apache.accumulo.core.trace.thrift.TInfo;
 import org.apache.accumulo.server.fs.VolumeManager;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
+import org.easymock.EasyMock;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
@@ -66,6 +73,7 @@ import org.junit.rules.TestName;
 
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Maps;
+import com.google.common.net.HostAndPort;
 
 public class GarbageCollectWriteAheadLogsTest {
   private static final long BLOCK_SIZE = 64000000L;
@@ -517,4 +525,75 @@ public class GarbageCollectWriteAheadLogsTest {
       }
     }
   }
+
+  @Test
+  public void getActiveWals() throws Exception {
+    GarbageCollectWriteAheadLogs gcWals = EasyMock.createMockBuilder(GarbageCollectWriteAheadLogs.class).addMockedMethod("getActiveTservers")
+        .addMockedMethod("getActiveWalsForServer").createMock();
+    AccumuloConfiguration conf = EasyMock.createMock(AccumuloConfiguration.class);
+    TInfo tinfo = EasyMock.createMock(TInfo.class);
+    TCredentials tcreds = EasyMock.createMock(TCredentials.class);
+
+    List<String> tservers = Arrays.asList("localhost:12345", "localhost:12346");
+    EasyMock.expect(gcWals.getActiveTservers(conf, tinfo, tcreds)).andReturn(tservers);
+    int numWals = 0;
+    for (String tserver : tservers) {
+      EasyMock.expect(gcWals.getActiveWalsForServer(conf, tinfo, tcreds, HostAndPort.fromString(tserver))).andReturn(Arrays.asList("/wal"
+ numWals));
+      numWals++;
+    }
+
+    EasyMock.replay(gcWals);
+
+    Set<String> wals = gcWals.getActiveWals(conf, tinfo, tcreds);
+
+    EasyMock.verify(gcWals);
+
+    Set<String> expectedWals = new HashSet<String>();
+    for (int i = 0; i < numWals; i++) {
+      expectedWals.add("/wal" + i);
+    }
+
+    Assert.assertEquals(expectedWals, wals);
+  }
+
+  @Test
+  public void offlineMaster() throws Exception {
+    GarbageCollectWriteAheadLogs gcWals = EasyMock.createMockBuilder(GarbageCollectWriteAheadLogs.class).addMockedMethod("getActiveTservers")
+        .addMockedMethod("getActiveWalsForServer").createMock();
+    AccumuloConfiguration conf = EasyMock.createMock(AccumuloConfiguration.class);
+    TInfo tinfo = EasyMock.createMock(TInfo.class);
+    TCredentials tcreds = EasyMock.createMock(TCredentials.class);
+
+    EasyMock.expect(gcWals.getActiveTservers(conf, tinfo, tcreds)).andReturn(null);
+
+    EasyMock.replay(gcWals);
+
+    Set<String> wals = gcWals.getActiveWals(conf, tinfo, tcreds);
+
+    EasyMock.verify(gcWals);
+
+    Assert.assertNull("Expected to get null for active WALs", wals);
+  }
+
+  @Test
+  public void offlineTserver() throws Exception {
+    GarbageCollectWriteAheadLogs gcWals = EasyMock.createMockBuilder(GarbageCollectWriteAheadLogs.class).addMockedMethod("getActiveTservers")
+        .addMockedMethod("getActiveWalsForServer").createMock();
+    AccumuloConfiguration conf = EasyMock.createMock(AccumuloConfiguration.class);
+    TInfo tinfo = EasyMock.createMock(TInfo.class);
+    TCredentials tcreds = EasyMock.createMock(TCredentials.class);
+
+    List<String> tservers = Arrays.asList("localhost:12345", "localhost:12346");
+    EasyMock.expect(gcWals.getActiveTservers(conf, tinfo, tcreds)).andReturn(tservers);
+    EasyMock.expect(gcWals.getActiveWalsForServer(conf, tinfo, tcreds, HostAndPort.fromString("localhost:12345"))).andReturn(Arrays.asList("/wal"
+ 0));
+    EasyMock.expect(gcWals.getActiveWalsForServer(conf, tinfo, tcreds, HostAndPort.fromString("localhost:12346"))).andReturn(null);
+
+    EasyMock.replay(gcWals);
+
+    Set<String> wals = gcWals.getActiveWals(conf, tinfo, tcreds);
+
+    EasyMock.verify(gcWals);
+
+    Assert.assertNull("Expected to get null for active WALs", wals);
+  }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/2b347346/server/master/src/main/java/org/apache/accumulo/master/MasterClientServiceHandler.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/MasterClientServiceHandler.java
b/server/master/src/main/java/org/apache/accumulo/master/MasterClientServiceHandler.java
index 2007b36..9a3e532 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/MasterClientServiceHandler.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/MasterClientServiceHandler.java
@@ -21,6 +21,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map.Entry;
 import java.util.Set;
 
@@ -435,7 +436,7 @@ class MasterClientServiceHandler extends FateServiceHandler implements
MasterCli
   private void alterTableProperty(TCredentials c, String tableName, String property, String
value, TableOperation op) throws ThriftSecurityException,
       ThriftTableOperationException {
     final String tableId = ClientServiceHandler.checkTableId(master.getInstance(), tableName,
op);
-    String namespaceId = Tables.getNamespaceId(master.getInstance(), tableId); 
+    String namespaceId = Tables.getNamespaceId(master.getInstance(), tableId);
     if (!master.security.canAlterTable(c, tableId, namespaceId))
       throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
 
@@ -472,4 +473,14 @@ class MasterClientServiceHandler extends FateServiceHandler implements
MasterCli
     master.waitForBalance(tinfo);
   }
 
+  @Override
+  public List<String> getActiveTservers(TInfo tinfo, TCredentials credentials) throws
TException {
+    Set<TServerInstance> tserverInstances = master.onlineTabletServers();
+    List<String> servers = new ArrayList<String>();
+    for (TServerInstance tserverInstance : tserverInstances) {
+      servers.add(tserverInstance.getLocation().toString());
+    }
+
+    return servers;
+  }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/2b347346/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 1e81947..6002789 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -1743,6 +1743,13 @@ public class TabletServer implements Runnable {
 
       return ret;
     }
+    
+    @Override
+    public List<String> getActiveLogs(TInfo tinfo, TCredentials credentials) throws
TException {
+      Set<String> logs = new HashSet<String>();
+      logger.getLogFiles(logs);
+      return new ArrayList<String>(logs);
+    }
   }
 
   private class SplitRunner implements Runnable {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/2b347346/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
b/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
index a47f21a..b6cf727 100644
--- a/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
+++ b/test/src/main/java/org/apache/accumulo/test/performance/thrift/NullTserver.java
@@ -224,6 +224,11 @@ public class NullTserver {
 
     @Override
     public void closeConditionalUpdate(TInfo tinfo, long sessID) throws TException {}
+    
+    @Override
+    public List<String> getActiveLogs(TInfo tinfo, TCredentials credentials) throws
TException {
+      return null;
+    }
   }
 
   static class Opts extends Help {


Mime
View raw message