Skip to content

Commit

Permalink
Merge pull request #1132 from metamx/uriSegmentLoaders
Browse files Browse the repository at this point in the history 
Overhaul of SegmentPullers to add consistency and retries
  • Loading branch information
@fjy
fjy committed Mar 30, 2015
2 parents 9c741d5 + 6d407e8 commit 1e38def
Show file tree
Hide file tree
Showing 30 changed files with 1,800 additions and 276 deletions.
12 changes: 11 additions & 1 deletion common/src/main/java/com/fasterxml/jackson/databind/introspect/GuiceInjectableValues.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@
import com.fasterxml.jackson.databind.InjectableValues;
import com.google.inject.Injector;
import com.google.inject.Key;
import com.metamx.common.IAE;

import java.lang.reflect.Type;

/**
*/
Expand All @@ -36,6 +39,13 @@ public Object findInjectableValue(
Object valueId, DeserializationContext ctxt, BeanProperty forProperty, Object beanInstance
)
{
return injector.getInstance((Key) valueId);
// From the docs: "Object that identifies value to inject; may be a simple name or more complex identifier object,
// whatever provider needs"
// Currently we should only be dealing with `Key` instances, and anything more advanced should be handled with
// great care
if(valueId instanceof Key){
return injector.getInstance((Key) valueId);
}
throw new IAE("Unknown class type [%s] for valueId [%s]", valueId.getClass().getCanonicalName(), valueId.toString());
}
}
128 changes: 77 additions & 51 deletions ...assandra-storage/src/main/java/io/druid/storage/cassandra/CassandraDataSegmentPuller.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,79 +17,105 @@

package io.druid.storage.cassandra;

import com.google.common.io.Files;
import com.google.common.base.Predicates;
import com.google.inject.Inject;
import com.metamx.common.CompressionUtils;
import com.metamx.common.ISE;
import com.metamx.common.RetryUtils;
import com.metamx.common.logger.Logger;
import com.netflix.astyanax.recipes.storage.ChunkedStorage;
import com.netflix.astyanax.recipes.storage.ObjectMetadata;
import io.druid.segment.loading.DataSegmentPuller;
import io.druid.segment.loading.SegmentLoadingException;
import io.druid.timeline.DataSegment;
import io.druid.utils.CompressionUtils;
import org.apache.commons.io.FileUtils;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.Callable;

/**
* Cassandra Segment Puller
*
* @author boneill42
*/
public class CassandraDataSegmentPuller extends CassandraStorage implements DataSegmentPuller
{
private static final Logger log = new Logger(CassandraDataSegmentPuller.class);
private static final int CONCURRENCY = 10;
private static final int BATCH_SIZE = 10;
private static final Logger log = new Logger(CassandraDataSegmentPuller.class);
private static final int CONCURRENCY = 10;
private static final int BATCH_SIZE = 10;

@Inject
public CassandraDataSegmentPuller(CassandraDataSegmentConfig config)
{
super(config);
}
public CassandraDataSegmentPuller(CassandraDataSegmentConfig config)
{
super(config);
}

@Override
public void getSegmentFiles(DataSegment segment, File outDir) throws SegmentLoadingException
{
String key = (String) segment.getLoadSpec().get("key");
log.info("Pulling index from C* at path[%s] to outDir[%s]", key, outDir);
@Override
public void getSegmentFiles(DataSegment segment, File outDir) throws SegmentLoadingException
{
String key = (String) segment.getLoadSpec().get("key");
getSegmentFiles(key, outDir);
}
public com.metamx.common.FileUtils.FileCopyResult getSegmentFiles(final String key, final File outDir) throws SegmentLoadingException{
log.info("Pulling index from C* at path[%s] to outDir[%s]", key, outDir);
if (!outDir.exists()) {
outDir.mkdirs();
}

if (!outDir.exists())
{
outDir.mkdirs();
}
if (!outDir.isDirectory()) {
throw new ISE("outDir[%s] must be a directory.", outDir);
}

if (!outDir.isDirectory())
{
throw new ISE("outDir[%s] must be a directory.", outDir);
}
long startTime = System.currentTimeMillis();
final File tmpFile = new File(outDir, "index.zip");
log.info("Pulling to temporary local cache [%s]", tmpFile.getAbsolutePath());

long startTime = System.currentTimeMillis();
ObjectMetadata meta = null;
final File outFile = new File(outDir, "index.zip");
try
{
try
{
log.info("Writing to [%s]", outFile.getAbsolutePath());
OutputStream os = Files.newOutputStreamSupplier(outFile).getOutput();
meta = ChunkedStorage
.newReader(indexStorage, key, os)
.withBatchSize(BATCH_SIZE)
.withConcurrencyLevel(CONCURRENCY)
.call();
os.close();
CompressionUtils.unzip(outFile, outDir);
} catch (Exception e)
{
FileUtils.deleteDirectory(outDir);
}
} catch (Exception e)
{
throw new SegmentLoadingException(e, e.getMessage());
}
log.info("Pull of file[%s] completed in %,d millis (%s bytes)", key, System.currentTimeMillis() - startTime,
meta.getObjectSize());
}
final com.metamx.common.FileUtils.FileCopyResult localResult;
try {
localResult = RetryUtils.retry(
new Callable<com.metamx.common.FileUtils.FileCopyResult>()
{
@Override
public com.metamx.common.FileUtils.FileCopyResult call() throws Exception
{
try (OutputStream os = new FileOutputStream(tmpFile)) {
final ObjectMetadata meta = ChunkedStorage
.newReader(indexStorage, key, os)
.withBatchSize(BATCH_SIZE)
.withConcurrencyLevel(CONCURRENCY)
.call();
}
return new com.metamx.common.FileUtils.FileCopyResult(tmpFile);
}
},
Predicates.<Throwable>alwaysTrue(),
10
);
}catch (Exception e){
throw new SegmentLoadingException(e, "Unable to copy key [%s] to file [%s]", key, tmpFile.getAbsolutePath());
}
try{
final com.metamx.common.FileUtils.FileCopyResult result = CompressionUtils.unzip(tmpFile, outDir);
log.info(
"Pull of file[%s] completed in %,d millis (%s bytes)", key, System.currentTimeMillis() - startTime,
result.size()
);
return result;
}
catch (Exception e) {
try {
FileUtils.deleteDirectory(outDir);
}
catch (IOException e1) {
log.error(e1, "Error clearing segment directory [%s]", outDir.getAbsolutePath());
e.addSuppressed(e1);
}
throw new SegmentLoadingException(e, e.getMessage());
} finally {
if(!tmpFile.delete()){
log.warn("Could not delete cache file at [%s]", tmpFile.getAbsolutePath());
}
}
}
}
2 changes: 1 addition & 1 deletion ...assandra-storage/src/main/java/io/druid/storage/cassandra/CassandraDataSegmentPusher.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,14 +21,14 @@
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.inject.Inject;
import com.metamx.common.CompressionUtils;
import com.metamx.common.logger.Logger;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.recipes.storage.ChunkedStorage;
import io.druid.segment.SegmentUtils;
import io.druid.segment.loading.DataSegmentPusher;
import io.druid.segment.loading.DataSegmentPusherUtil;
import io.druid.timeline.DataSegment;
import io.druid.utils.CompressionUtils;

import java.io.File;
import java.io.FileInputStream;
Expand Down
43 changes: 33 additions & 10 deletions ...ions/cassandra-storage/src/main/java/io/druid/storage/cassandra/CassandraDruidModule.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package io.druid.storage.cassandra;

import com.fasterxml.jackson.databind.Module;
import com.fasterxml.jackson.core.Version;
import com.google.common.collect.ImmutableList;
import com.google.inject.Binder;
import com.google.inject.Key;
Expand All @@ -34,24 +34,47 @@
*/
public class CassandraDruidModule implements DruidModule
{
@Override
public List<? extends Module> getJacksonModules()
{
return ImmutableList.of();
}
public static final String SCHEME = "c*";

@Override
public void configure(Binder binder)
{
Binders.dataSegmentPullerBinder(binder)
.addBinding("c*")
.to(CassandraDataSegmentPuller.class)
.in(LazySingleton.class);
.addBinding(SCHEME)
.to(CassandraDataSegmentPuller.class)
.in(LazySingleton.class);

PolyBind.optionBinder(binder, Key.get(DataSegmentPusher.class))
.addBinding("c*")
.addBinding(SCHEME)
.to(CassandraDataSegmentPusher.class)
.in(LazySingleton.class);
JsonConfigProvider.bind(binder, "druid.storage", CassandraDataSegmentConfig.class);
}

@Override
public List<? extends com.fasterxml.jackson.databind.Module> getJacksonModules()
{
return ImmutableList.of(
new com.fasterxml.jackson.databind.Module()
{
@Override
public String getModuleName()
{
return "DruidCassandraStorage-" + System.identityHashCode(this);
}

@Override
public Version version()
{
return Version.unknownVersion();
}

@Override
public void setupModule(SetupContext context)
{
context.registerSubtypes(CassandraLoadSpec.class);
}
}
);
}
}
54 changes: 54 additions & 0 deletions extensions/cassandra-storage/src/main/java/io/druid/storage/cassandra/CassandraLoadSpec.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
/*
* Druid - a distributed column store.
* Copyright 2012 - 2015 Metamarkets Group Inc.
*
* Licensed 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 io.druid.storage.cassandra;

import com.fasterxml.jackson.annotation.JacksonInject;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
import io.druid.segment.loading.LoadSpec;
import io.druid.segment.loading.SegmentLoadingException;

import java.io.File;

/**
*
*/
@JsonTypeName(CassandraDruidModule.SCHEME)
public class CassandraLoadSpec implements LoadSpec
{
@JsonProperty
private final String key;
private final CassandraDataSegmentPuller puller;

@JsonCreator
public CassandraLoadSpec(
@JacksonInject CassandraDataSegmentPuller puller,
@JsonProperty("key") String key
)
{
this.puller = puller;
this.key = key;
}

@Override
public LoadSpecResult loadSegment(File outDir) throws SegmentLoadingException
{
return new LoadSpecResult(puller.getSegmentFiles(key, outDir).size());
}
}
36 changes: 31 additions & 5 deletions extensions/hdfs-storage/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,13 +63,39 @@
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>

<!-- Tests -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.druid</groupId>
<artifactId>druid-server</artifactId>
<version>${parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.3.0</version>
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.3.0</version>
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.3.0</version>
<scope>test</scope>
</dependency>
</dependencies>
</dependencies>

</project>
Loading

0 comments on commit 1e38def

Please sign in to comment.