From 6b903720021683c7d12b5befee9656fc4edea7ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Xavier=20L=C3=A9aut=C3=A9?= Date: Fri, 13 Dec 2013 16:40:57 -0800 Subject: [PATCH] separate segment mover and segment archiver --- pom.xml | 2 +- .../storage/s3/S3DataSegmentArchiver.java | 59 +++++++++++++++++++ ....java => S3DataSegmentArchiverConfig.java} | 2 +- .../druid/storage/s3/S3DataSegmentMover.java | 34 ++++++----- .../storage/s3/S3StorageDruidModule.java | 3 +- 5 files changed, 82 insertions(+), 18 deletions(-) create mode 100644 s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentArchiver.java rename s3-extensions/src/main/java/io/druid/storage/s3/{S3DataSegmentMoverConfig.java => S3DataSegmentArchiverConfig.java} (96%) diff --git a/pom.xml b/pom.xml index f970d49dea..f55d5202e2 100644 --- a/pom.xml +++ b/pom.xml @@ -41,7 +41,7 @@ UTF-8 0.25.1 2.1.0-incubating - 0.1.6 + 0.1.7 diff --git a/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentArchiver.java b/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentArchiver.java new file mode 100644 index 0000000000..b339187ff2 --- /dev/null +++ b/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentArchiver.java @@ -0,0 +1,59 @@ +/* + * Druid - a distributed column store. + * Copyright (C) 2012, 2013 Metamarkets Group Inc. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version 2 + * of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + */ + +package io.druid.storage.s3; + +import com.google.common.collect.ImmutableMap; +import com.google.inject.Inject; +import com.metamx.common.MapUtils; +import io.druid.segment.loading.DataSegmentArchiver; +import io.druid.segment.loading.SegmentLoadingException; +import io.druid.timeline.DataSegment; +import org.jets3t.service.impl.rest.httpclient.RestS3Service; + + +public class S3DataSegmentArchiver extends S3DataSegmentMover implements DataSegmentArchiver +{ + private final S3DataSegmentArchiverConfig config; + + @Inject + public S3DataSegmentArchiver( + RestS3Service s3Client, + S3DataSegmentArchiverConfig config + ) + { + super(s3Client); + this.config = config; + } + + @Override + public DataSegment archive(DataSegment segment) throws SegmentLoadingException + { + String targetS3Bucket = config.getArchiveBucket(); + String targetS3Path = MapUtils.getString(segment.getLoadSpec(), "key"); + + return move( + segment, + ImmutableMap.of( + "bucket", targetS3Bucket, + "key", targetS3Path + ) + ); + } +} diff --git a/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentMoverConfig.java b/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentArchiverConfig.java similarity index 96% rename from s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentMoverConfig.java rename to s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentArchiverConfig.java index a298ba9da3..53a04e4310 100644 --- a/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentMoverConfig.java +++ b/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentArchiverConfig.java @@ -21,7 +21,7 @@ package io.druid.storage.s3; import com.fasterxml.jackson.annotation.JsonProperty; -public class S3DataSegmentMoverConfig +public class S3DataSegmentArchiverConfig { @JsonProperty public String archiveBucket = ""; diff --git a/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentMover.java b/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentMover.java index 9c79a07980..9b3f122b59 100644 --- a/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentMover.java +++ b/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentMover.java @@ -37,20 +37,17 @@ public class S3DataSegmentMover implements DataSegmentMover private static final Logger log = new Logger(S3DataSegmentMover.class); private final RestS3Service s3Client; - private final S3DataSegmentMoverConfig config; @Inject public S3DataSegmentMover( - RestS3Service s3Client, - S3DataSegmentMoverConfig config + RestS3Service s3Client ) { this.s3Client = s3Client; - this.config = config; } @Override - public DataSegment move(DataSegment segment) throws SegmentLoadingException + public DataSegment move(DataSegment segment, Map targetLoadSpec) throws SegmentLoadingException { try { Map loadSpec = segment.getLoadSpec(); @@ -58,10 +55,15 @@ public class S3DataSegmentMover implements DataSegmentMover String s3Path = MapUtils.getString(loadSpec, "key"); String s3DescriptorPath = S3Utils.descriptorPathForSegmentPath(s3Path); - final String s3ArchiveBucket = config.getArchiveBucket(); + final String targetS3Bucket = MapUtils.getString(targetLoadSpec, "bucket"); + final String targetS3Path = MapUtils.getString(targetLoadSpec, "key"); + String targetS3DescriptorPath = S3Utils.descriptorPathForSegmentPath(targetS3Path); - if (s3ArchiveBucket.isEmpty()) { - throw new SegmentLoadingException("S3 archive bucket not specified"); + if (targetS3Bucket.isEmpty()) { + throw new SegmentLoadingException("Target S3 bucket is not specified"); + } + if (targetS3Path.isEmpty()) { + throw new SegmentLoadingException("Target S3 path is not specified"); } if (s3Client.isObjectInBucket(s3Bucket, s3Path)) { @@ -69,26 +71,28 @@ public class S3DataSegmentMover implements DataSegmentMover "Moving index file[s3://%s/%s] to [s3://%s/%s]", s3Bucket, s3Path, - s3ArchiveBucket, - s3Path + targetS3Bucket, + targetS3Path ); - s3Client.moveObject(s3Bucket, s3Path, s3ArchiveBucket, new S3Object(s3Path), false); + s3Client.moveObject(s3Bucket, s3Path, targetS3Bucket, new S3Object(targetS3Path), false); } if (s3Client.isObjectInBucket(s3Bucket, s3DescriptorPath)) { log.info( "Moving descriptor file[s3://%s/%s] to [s3://%s/%s]", s3Bucket, s3DescriptorPath, - s3ArchiveBucket, - s3DescriptorPath + targetS3Bucket, + targetS3DescriptorPath ); - s3Client.moveObject(s3Bucket, s3DescriptorPath, s3ArchiveBucket, new S3Object(s3DescriptorPath), false); + s3Client.moveObject(s3Bucket, s3DescriptorPath, targetS3Bucket, new S3Object(targetS3DescriptorPath), false); } return segment.withLoadSpec( ImmutableMap.builder() .putAll(loadSpec) - .put("bucket", s3ArchiveBucket).build() + .put("bucket", targetS3Bucket) + .put("key", targetS3Path) + .build() ); } catch (ServiceException e) { diff --git a/s3-extensions/src/main/java/io/druid/storage/s3/S3StorageDruidModule.java b/s3-extensions/src/main/java/io/druid/storage/s3/S3StorageDruidModule.java index fadba584bc..d30f49f976 100644 --- a/s3-extensions/src/main/java/io/druid/storage/s3/S3StorageDruidModule.java +++ b/s3-extensions/src/main/java/io/druid/storage/s3/S3StorageDruidModule.java @@ -52,9 +52,10 @@ public class S3StorageDruidModule implements DruidModule Binders.dataSegmentPullerBinder(binder).addBinding("s3_zip").to(S3DataSegmentPuller.class).in(LazySingleton.class); Binders.dataSegmentKillerBinder(binder).addBinding("s3_zip").to(S3DataSegmentKiller.class).in(LazySingleton.class); Binders.dataSegmentMoverBinder(binder).addBinding("s3_zip").to(S3DataSegmentMover.class).in(LazySingleton.class); + Binders.dataSegmentArchiverBinder(binder).addBinding("s3_zip").to(S3DataSegmentArchiver.class).in(LazySingleton.class); Binders.dataSegmentPusherBinder(binder).addBinding("s3").to(S3DataSegmentPusher.class).in(LazySingleton.class); JsonConfigProvider.bind(binder, "druid.storage", S3DataSegmentPusherConfig.class); - JsonConfigProvider.bind(binder, "druid.storage", S3DataSegmentMoverConfig.class); + JsonConfigProvider.bind(binder, "druid.storage", S3DataSegmentArchiverConfig.class); Binders.taskLogsBinder(binder).addBinding("s3").to(S3TaskLogs.class); JsonConfigProvider.bind(binder, "druid.indexer.logs", S3TaskLogsConfig.class); -- GitLab