AMBARI-22731. Infra Manager - hdfs upload support for archiving Infra Solr
diff --git a/ambari-infra-manager-it/pom.xml b/ambari-infra-manager-it/pom.xml
index 97e8ea0..6dcb4b6 100644
--- a/ambari-infra-manager-it/pom.xml
+++ b/ambari-infra-manager-it/pom.xml
@@ -89,6 +89,18 @@
<version>1.3</version>
<scope>test</scope>
</dependency>
+ <!-- Hadoop -->
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs-client</artifactId>
+ <version>${hadoop.version}</version>
+ </dependency>
</dependencies>
<build>
diff --git a/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/InfraClient.java b/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/InfraClient.java
index 0e391a3..b798ce1 100644
--- a/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/InfraClient.java
+++ b/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/InfraClient.java
@@ -18,6 +18,10 @@
*/
package org.apache.ambari.infra;
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.io.IOUtils;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.CloseableHttpResponse;
@@ -36,6 +40,8 @@
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.Charset;
+import java.util.HashMap;
+import java.util.Map;
import static org.apache.commons.lang.StringUtils.isBlank;
@@ -77,14 +83,29 @@
}
}
- // TODO: return job data
- public void startJob(String jobName, String parameters) {
+ public String startJob(String jobName, String parameters) {
URIBuilder uriBuilder = new URIBuilder(baseUrl);
uriBuilder.setScheme("http");
uriBuilder.setPath(uriBuilder.getPath() + "/" + jobName);
if (!isBlank(parameters))
uriBuilder.addParameter("params", parameters);
try {
+ String responseText = execute(new HttpPost(uriBuilder.build()));
+ Map<String, Object> responseContent = new ObjectMapper().readValue(responseText, new TypeReference<HashMap<String,Object>>() {});
+ return responseContent.get("jobId").toString();
+ } catch (URISyntaxException | JsonParseException | JsonMappingException e) {
+ throw new RuntimeException(e);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public void restartJob(String jobName, String jobId) {
+ URIBuilder uriBuilder = new URIBuilder(baseUrl);
+ uriBuilder.setScheme("http");
+ uriBuilder.setPath(String.format("%s/%s/%s/executions", uriBuilder.getPath(), jobName, jobId));
+ uriBuilder.addParameter("operation", "RESTART");
+ try {
execute(new HttpPost(uriBuilder.build()));
} catch (URISyntaxException e) {
throw new RuntimeException(e);
diff --git a/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/steps/AbstractInfraSteps.java b/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/steps/AbstractInfraSteps.java
index f48d4c2..ece1c59 100644
--- a/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/steps/AbstractInfraSteps.java
+++ b/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/steps/AbstractInfraSteps.java
@@ -23,8 +23,14 @@
import com.amazonaws.services.s3.model.ListObjectsRequest;
import com.amazonaws.services.s3.model.ObjectListing;
import org.apache.ambari.infra.InfraClient;
+import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
@@ -56,6 +62,7 @@
private static final int SOLR_PORT = 8983;
private static final int INFRA_MANAGER_PORT = 61890;
private static final int FAKE_S3_PORT = 4569;
+ private static final int HDFS_PORT = 9000;
private static final String AUDIT_LOGS_COLLECTION = "audit_logs";
protected static final String S3_BUCKET_NAME = "testbucket";
private String ambariFolder;
@@ -77,13 +84,22 @@
return s3client;
}
+ public String getLocalDataFolder() {
+ return ambariFolder + "/ambari-infra/ambari-infra-manager/docker/test-out";
+ }
+
@BeforeStories
public void initDockerContainer() throws Exception {
- LOG.info("Create new docker container for testing Ambari Infra Manager ...");
+ System.setProperty("HADOOP_USER_NAME", "root");
+
URL location = AbstractInfraSteps.class.getProtectionDomain().getCodeSource().getLocation();
ambariFolder = new File(location.toURI()).getParentFile().getParentFile().getParentFile().getParent();
- shellScriptLocation = ambariFolder + "/ambari-infra/ambari-infra-manager/docker/infra-manager-docker-compose.sh";
+ LOG.info("Clean local data folder {}", getLocalDataFolder());
+ FileUtils.cleanDirectory(new File(getLocalDataFolder()));
+
+ shellScriptLocation = ambariFolder + "/ambari-infra/ambari-infra-manager/docker/infra-manager-docker-compose.sh";
+ LOG.info("Create new docker container for testing Ambari Infra Manager ...");
runCommand(new String[]{shellScriptLocation, "start"});
dockerHost = System.getProperty("docker.host") != null ? System.getProperty("docker.host") : "localhost";
@@ -106,7 +122,7 @@
checkInfraManagerReachable();
}
- protected void runCommand(String[] command) {
+ private void runCommand(String[] command) {
try {
LOG.info("Exec command: {}", StringUtils.join(command, " "));
Process process = Runtime.getRuntime().exec(command);
@@ -130,7 +146,7 @@
});
}
- protected void doWithin(int sec, String actionName, Runnable runnable) {
+ private void doWithin(int sec, String actionName, Runnable runnable) {
long start = currentTimeMillis();
Exception exception;
while (true) {
@@ -215,9 +231,26 @@
ListObjectsRequest listObjectsRequest = new ListObjectsRequest().withBucketName(S3_BUCKET_NAME);
ObjectListing objectListing = getS3client().listObjects(listObjectsRequest);
LOG.info("Found {} files on s3.", objectListing.getObjectSummaries().size());
- objectListing.getObjectSummaries().forEach(s3ObjectSummary -> LOG.info("Found file in s3 with key {}", s3ObjectSummary.getKey()));
+ objectListing.getObjectSummaries().forEach(s3ObjectSummary -> LOG.info("Found file on s3 with key {}", s3ObjectSummary.getKey()));
+
+ LOG.info("Listing files on hdfs.");
+ try (FileSystem fileSystem = getHdfs()) {
+ int count = 0;
+ RemoteIterator<LocatedFileStatus> it = fileSystem.listFiles(new Path("/test_audit_logs"), true);
+ while (it.hasNext()) {
+ LOG.info("Found file on hdfs with name {}", it.next().getPath().getName());
+ ++count;
+ }
+ LOG.info("{} files found on hfds", count);
+ }
LOG.info("shutdown containers");
runCommand(new String[]{shellScriptLocation, "stop"});
}
+
+ protected FileSystem getHdfs() throws IOException {
+ Configuration conf = new Configuration();
+ conf.set("fs.defaultFS", String.format("hdfs://%s:%d/", dockerHost, HDFS_PORT));
+ return FileSystem.get(conf);
+ }
}
diff --git a/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/steps/ExportJobsSteps.java b/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/steps/ExportJobsSteps.java
index 22826a0..7e54a31 100644
--- a/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/steps/ExportJobsSteps.java
+++ b/ambari-infra-manager-it/src/test/java/org/apache/ambari/infra/steps/ExportJobsSteps.java
@@ -23,9 +23,12 @@
import com.amazonaws.services.s3.model.ObjectListing;
import com.amazonaws.services.s3.model.ObjectMetadata;
import org.apache.ambari.infra.InfraClient;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
-import org.jbehave.core.annotations.Alias;
import org.jbehave.core.annotations.Given;
import org.jbehave.core.annotations.Then;
import org.jbehave.core.annotations.When;
@@ -33,11 +36,16 @@
import org.slf4j.LoggerFactory;
import java.io.ByteArrayInputStream;
+import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.time.Duration;
import java.time.OffsetDateTime;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import static java.util.Objects.requireNonNull;
import static org.apache.ambari.infra.OffsetDateTimeConverter.SOLR_DATETIME_FORMATTER;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.Matchers.containsString;
@@ -48,6 +56,8 @@
public class ExportJobsSteps extends AbstractInfraSteps {
private static final Logger LOG = LoggerFactory.getLogger(ExportJobsSteps.class);
+ private Map<String, String> launchedJobs = new HashMap<>();
+
@Given("$count documents in solr")
public void addDocuments(int count) throws Exception {
OffsetDateTime intervalEnd = OffsetDateTime.now();
@@ -79,10 +89,18 @@
}
@When("start $jobName job with parameters $parameters")
- @Alias("restart $jobName job with parameters $parameters")
public void startJob(String jobName, String parameters) throws Exception {
try (InfraClient httpClient = getInfraClient()) {
- httpClient.startJob(jobName, parameters);
+ String jobId = httpClient.startJob(jobName, parameters);
+ LOG.info("Job {} started jobId: {}", jobName, jobId);
+ launchedJobs.put(jobName, jobId);
+ }
+ }
+
+ @When("restart $jobName job")
+ public void restartJob(String jobName) throws Exception {
+ try (InfraClient httpClient = getInfraClient()) {
+ httpClient.restartJob(jobName, launchedJobs.get(jobName));
}
}
@@ -103,7 +121,7 @@
}
@Then("Check $count files exists on s3 server with filenames containing the text $text after $waitSec seconds")
- public void checkNumberOfFilesOnS3(int count, String text, int waitSec) {
+ public void checkNumberOfFilesOnS3(long count, String text, int waitSec) {
AmazonS3Client s3Client = getS3client();
ListObjectsRequest listObjectsRequest = new ListObjectsRequest().withBucketName(S3_BUCKET_NAME);
doWithin(waitSec, "check uploaded files to s3", () -> s3Client.doesBucketExist(S3_BUCKET_NAME)
@@ -145,4 +163,37 @@
throw new UncheckedIOException(e);
}
}
+
+ @Then("Check $count files exists on hdfs with filenames containing the text $text in the folder $path after $waitSec seconds")
+ public void checkNumberOfFilesOnHdfs(int count, String text, String path, int waitSec) throws Exception {
+ try (FileSystem fileSystem = getHdfs()) {
+ doWithin(waitSec, "check uploaded files to hdfs", () -> {
+ try {
+ int fileCount = 0;
+ RemoteIterator<LocatedFileStatus> it = fileSystem.listFiles(new Path(path), true);
+ while (it.hasNext()) {
+ if (it.next().getPath().getName().contains(text))
+ ++fileCount;
+ }
+ return fileCount == count;
+ }
+ catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ }
+ }
+
+ @Then("Check $count files exists on local filesystem with filenames containing the text $text in the folder $path")
+ public void checkNumberOfFilesOnLocalFilesystem(long count, String text, String path) {
+ File destinationDirectory = new File(getLocalDataFolder(), path);
+ LOG.info("Destination directory path: {}", destinationDirectory.getAbsolutePath());
+ doWithin(5, "Destination directory exists", destinationDirectory::exists);
+
+ File[] files = requireNonNull(destinationDirectory.listFiles(),
+ String.format("Path %s is not a directory or an I/O error occurred!", destinationDirectory.getAbsolutePath()));
+ assertThat(Arrays.stream(files)
+ .filter(file -> file.getName().contains(text))
+ .count(), is(count));
+ }
}
diff --git a/ambari-infra-manager-it/src/test/resources/stories/infra_api_tests.story b/ambari-infra-manager-it/src/test/resources/stories/infra_api_tests.story
index 1044258..d1eb4a4 100644
--- a/ambari-infra-manager-it/src/test/resources/stories/infra_api_tests.story
+++ b/ambari-infra-manager-it/src/test/resources/stories/infra_api_tests.story
@@ -12,6 +12,7 @@
Then Check 4 files exists on s3 server with filenames containing the text solr_archive_audit_logs_-_2010-10-09 after 20 seconds
And solr does not contain documents between 2010-10-09T05:00:00.000Z and 2010-10-09T20:00:00.000Z after 5 seconds
+
Scenario: Running archiving job with a bigger start value than end value exports and deletes 0 documents
Given 10 documents in solr with logtime from 2010-01-01T05:00:00.000Z to 2010-01-04T05:00:00.000Z
@@ -19,6 +20,7 @@
Then No file exists on s3 server with filenames containing the text solr_archive_audit_logs_-_2010-01-0
And solr contains 10 documents between 2010-01-01T05:00:00.000Z and 2010-01-04T05:00:00.000Z
+
Scenario: Archiving job fails when part of the data is exported. After resolving the issue and restarting the job exports the rest of the data.
Given 200 documents in solr with logtime from 2011-10-09T05:00:00.000Z to 2011-10-09T20:00:00.000Z
@@ -27,12 +29,29 @@
Then Check 3 files exists on s3 server with filenames containing the text solr_archive_audit_logs_-_2011-10-09 after 20 seconds
And solr does not contain documents between 2011-10-09T05:00:00.000Z and 2011-10-09T07:59:59.999Z after 5 seconds
When delete file with key solr_archive_audit_logs_-_2011-10-09T08-00-00.000Z.json.tar.gz from s3
-And restart archive_audit_logs job with parameters writeBlockSize=20,start=2010-11-09T00:00:00.000Z,end=2011-10-11T00:00:00.000Z
+And restart archive_audit_logs job
Then Check 10 files exists on s3 server with filenames containing the text solr_archive_audit_logs_-_2011-10-09 after 20 seconds
And solr does not contain documents between 2011-10-09T05:00:00.000Z and 2011-10-09T20:00:00.000Z after 5 seconds
+
Scenario: After Deleting job deletes documents from solr no document found in the specified interval
Given 10 documents in solr with logtime from 2012-10-09T05:00:00.000Z to 2012-10-09T20:00:00.000Z
When start delete_audit_logs job with parameters start=2012-10-09T05:00:00.000Z,end=2012-10-09T20:00:00.000Z
-Then solr does not contain documents between 2012-10-09T05:00:00.000Z and 2012-10-09T20:00:00.000Z after 5 seconds
\ No newline at end of file
+Then solr does not contain documents between 2012-10-09T05:00:00.000Z and 2012-10-09T20:00:00.000Z after 5 seconds
+
+
+Scenario: Archiving documents to hdfs
+
+Given 1000 documents in solr with logtime from 2014-01-04T05:00:00.000Z to 2014-01-06T20:00:00.000Z
+When start archive_audit_logs job with parameters start=2014-01-04T05:00:00.000Z,end=2014-01-06T20:00:00.000Z,destination=HDFS
+Then Check 7 files exists on hdfs with filenames containing the text audit_logs_-_2014-01-0 in the folder /test_audit_logs after 10 seconds
+And solr does not contain documents between 2014-01-04T05:00:00.000Z and 2014-01-06T20:00:00.000Z after 10 seconds
+
+
+Scenario: Archiving documents to local filesystem
+
+Given 200 documents in solr with logtime from 2014-02-04T05:00:00.000Z to 2014-02-06T20:00:00.000Z
+When start archive_audit_logs job with parameters start=2014-02-04T05:00:00.000Z,end=2014-02-06T20:00:00.000Z,destination=LOCAL,localDestinationDirectory=/root/archive
+Then Check 2 files exists on local filesystem with filenames containing the text audit_logs_-_2014-02-0 in the folder audit_logs_8_2014-02-06T20-00-00.000Z
+And solr does not contain documents between 2014-02-04T05:00:00.000Z and 2014-02-06T20:00:00.000Z after 10 seconds
diff --git a/ambari-infra-manager/.gitignore b/ambari-infra-manager/.gitignore
index ba4e51d..94b3829 100644
--- a/ambari-infra-manager/.gitignore
+++ b/ambari-infra-manager/.gitignore
@@ -1,2 +1,5 @@
out/*
-*.pid
\ No newline at end of file
+*.pid
+Profile
+.env
+test-out
\ No newline at end of file
diff --git a/ambari-infra-manager/docker/docker-compose.yml b/ambari-infra-manager/docker/docker-compose.yml
index 1172631..4f1febd 100644
--- a/ambari-infra-manager/docker/docker-compose.yml
+++ b/ambari-infra-manager/docker/docker-compose.yml
@@ -45,19 +45,39 @@
- ${ZOOKEEPER_CONNECTION_STRING}
volumes:
- $AMBARI_LOCATION/ambari-logsearch/ambari-logsearch-server/src/main/configsets:/opt/solr/configsets
- localstack-s3:
+ fakes3:
image: localstack/localstack
+ hostname: fakes3
ports:
- "4569:4569"
environment:
- SERVICES=s3:4569
- hostname: fakes3
networks:
infra-network:
aliases:
- testbucket.fakes3
env_file:
- Profile
+ namenode:
+ image: flokkr/hadoop-hdfs-namenode:${HADOOP_VERSION:-3.0.0}
+ hostname: namenode
+ ports:
+ - 9870:9870
+ - 9000:9000
+ env_file:
+ - Profile
+ environment:
+ ENSURE_NAMENODE_DIR: "/tmp/hadoop-hdfs/dfs/name"
+ networks:
+ - infra-network
+ datanode:
+ image: flokkr/hadoop-hdfs-datanode:${HADOOP_VERSION:-3.0.0}
+ links:
+ - namenode
+ env_file:
+ - Profile
+ networks:
+ - infra-network
inframanager:
image: ambari-infra-manager:v1.0
restart: always
@@ -76,6 +96,7 @@
DISPLAY: $DOCKERIP:0
volumes:
- $AMBARI_LOCATION/ambari-infra/ambari-infra-manager/target/package:/root/ambari-infra-manager
+ - $AMBARI_LOCATION/ambari-infra/ambari-infra-manager/docker/test-out:/root/archive
networks:
infra-network:
driver: bridge
diff --git a/ambari-infra-manager/docker/infra-manager-docker-compose.sh b/ambari-infra-manager/docker/infra-manager-docker-compose.sh
index 6ddb7c2..e5df48c 100755
--- a/ambari-infra-manager/docker/infra-manager-docker-compose.sh
+++ b/ambari-infra-manager/docker/infra-manager-docker-compose.sh
@@ -20,12 +20,12 @@
function start_containers() {
check_env_files
- echo "Start containers ..."
+ kill_containers
pushd $sdir/../
local AMBARI_INFRA_MANAGER_LOCATION=$(pwd)
echo $AMBARI_INFRA_MANAGER_LOCATION
- kill_containers
cd $AMBARI_INFRA_MANAGER_LOCATION/docker
+ echo "Start containers ..."
docker-compose up -d
popd
echo "Containers started"
@@ -73,25 +73,37 @@
ZOOKEEPER_CONNECTION_STRING=zookeeper:2181
SOLR_VERSION=6.6.2
+
+HADOOP_VERSION=3.0.0
EOF
}
function setup_profile() {
- pushd $sdir/../../
- local AMBARI_LOCATION=$(pwd)
- popd
cat << EOF > $sdir/Profile
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
+HADOOP_USER_NAME=root
+
+CORE-SITE.XML_fs.default.name=hdfs://namenode:9000
+CORE-SITE.XML_fs.defaultFS=hdfs://namenode:9000
+HDFS-SITE.XML_dfs.namenode.rpc-address=namenode:9000
+HDFS-SITE.XML_dfs.replication=1
EOF
}
function kill_containers() {
+ pushd $sdir/../
+ local AMBARI_INFRA_MANAGER_LOCATION=$(pwd)
echo "Try to remove containers if exists ..."
- docker rm -f docker_inframanager_1
- docker rm -f docker_solr_1
- docker rm -f docker_zookeeper_1
- docker rm -f docker_localstack-s3_1
+ echo $AMBARI_INFRA_MANAGER_LOCATION
+ cd $AMBARI_INFRA_MANAGER_LOCATION/docker
+ docker-compose rm -f -s inframanager
+ docker-compose rm -f -s solr
+ docker-compose rm -f -s zookeeper
+ docker-compose rm -f -s fakes3
+ docker-compose rm -f -s namenode
+ docker-compose rm -f -s datanode
+ popd
}
case $command in
diff --git a/ambari-infra-manager/pom.xml b/ambari-infra-manager/pom.xml
index de131b0..75adb8d 100644
--- a/ambari-infra-manager/pom.xml
+++ b/ambari-infra-manager/pom.xml
@@ -277,7 +277,7 @@
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
- <version>2.7.0</version>
+ <version>${hadoop.version}</version>
<exclusions>
<exclusion>
<groupId>javax.servlet</groupId>
@@ -311,6 +311,33 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-server</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-util</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-servlet</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-security</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs-client</artifactId>
+ <version>${hadoop.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </exclusion>
</exclusions>
</dependency>
<dependency>
diff --git a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/JobProperties.java b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/JobProperties.java
index 0841dd7..292e15e 100644
--- a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/JobProperties.java
+++ b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/JobProperties.java
@@ -18,7 +18,7 @@
*/
package org.apache.ambari.infra.job;
-import org.apache.htrace.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.batch.core.JobParameters;
import java.io.IOException;
diff --git a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentArchivingConfiguration.java b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentArchivingConfiguration.java
index 7588b99..4fdc5da 100644
--- a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentArchivingConfiguration.java
+++ b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentArchivingConfiguration.java
@@ -18,8 +18,10 @@
*/
package org.apache.ambari.infra.job.archive;
+import org.apache.ambari.infra.conf.InfraManagerDataConfig;
import org.apache.ambari.infra.job.JobPropertyMap;
import org.apache.ambari.infra.job.ObjectSource;
+import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
@@ -37,11 +39,13 @@
import javax.annotation.PostConstruct;
import javax.inject.Inject;
import java.io.File;
-import java.nio.file.Paths;
+
+import static org.apache.commons.lang.StringUtils.isBlank;
@Configuration
public class DocumentArchivingConfiguration {
private static final Logger LOG = LoggerFactory.getLogger(DocumentArchivingConfiguration.class);
+ private static final DocumentWiper NOT_DELETE = (firstDocument, lastDocument) -> { };
@Inject
private DocumentExportPropertyMap propertyMap;
@@ -62,6 +66,9 @@
@PostConstruct
public void createJobs() {
+ if (propertyMap == null || propertyMap.getSolrDataExport() == null)
+ return;
+
propertyMap.getSolrDataExport().values().forEach(DocumentExportProperties::validate);
propertyMap.getSolrDataExport().keySet().forEach(jobName -> {
@@ -88,32 +95,59 @@
public DocumentExporter documentExporter(DocumentItemReader documentItemReader,
@Value("#{stepExecution.jobExecution.id}") String jobId,
@Value("#{stepExecution.jobExecution.executionContext.get('jobProperties')}") DocumentExportProperties properties,
- SolrDAO solrDAO) {
- File path = Paths.get(
- properties.getDestinationDirectoryPath(),
- // TODO: jobId should remain the same after continuing job
- String.format("%s_%s", properties.getSolr().getCollection(), jobId)).toFile(); // TODO: add end date
- LOG.info("Destination directory path={}", path);
- if (!path.exists()) {
- if (!path.mkdirs()) {
- LOG.warn("Unable to create directory {}", path);
- }
+ InfraManagerDataConfig infraManagerDataConfig,
+ @Value("#{jobParameters[end]}") String intervalEnd,
+ DocumentWiper documentWiper) {
+
+ File baseDir = new File(infraManagerDataConfig.getDataFolder(), "exporting");
+ CompositeFileAction fileAction = new CompositeFileAction(new TarGzCompressor());
+ switch (properties.getDestination()) {
+ case S3:
+ fileAction.add(new S3Uploader(properties.s3Properties().orElseThrow(() -> new IllegalStateException("S3 properties are not provided!"))));
+ break;
+ case HDFS:
+ org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
+ conf.set("fs.defaultFS", properties.getHdfsEndpoint());
+ fileAction.add(new HdfsUploader(conf, new Path(properties.getHdfsDestinationDirectory())));
+ break;
+ case LOCAL:
+ baseDir = new File(properties.getLocalDestinationDirectory());
+ break;
}
- CompositeFileAction fileAction = new CompositeFileAction(new TarGzCompressor());
- properties.s3Properties().ifPresent(s3Properties -> fileAction.add(new S3Uploader(s3Properties)));
FileNameSuffixFormatter fileNameSuffixFormatter = FileNameSuffixFormatter.from(properties);
- LocalItemWriterListener itemWriterListener = new LocalItemWriterListener(fileAction, solrDAO);
+ LocalItemWriterListener itemWriterListener = new LocalItemWriterListener(fileAction, documentWiper);
+ File destinationDirectory = new File(
+ baseDir,
+ String.format("%s_%s_%s",
+ properties.getSolr().getCollection(),
+ jobId,
+ isBlank(intervalEnd) ? "" : fileNameSuffixFormatter.format(intervalEnd)));
+ LOG.info("Destination directory path={}", destinationDirectory);
+ if (!destinationDirectory.exists()) {
+ if (!destinationDirectory.mkdirs()) {
+ LOG.warn("Unable to create directory {}", destinationDirectory);
+ }
+ }
return new DocumentExporter(
documentItemReader,
firstDocument -> new LocalDocumentItemWriter(
- outFile(properties.getSolr().getCollection(), path, fileNameSuffixFormatter.format(firstDocument)), itemWriterListener),
+ outFile(properties.getSolr().getCollection(), destinationDirectory, fileNameSuffixFormatter.format(firstDocument)), itemWriterListener),
properties.getWriteBlockSize());
}
@Bean
@StepScope
+ public DocumentWiper documentWiper(@Value("#{stepExecution.jobExecution.executionContext.get('jobProperties')}") DocumentExportProperties properties,
+ SolrDAO solrDAO) {
+ if (isBlank(properties.getSolr().getDeleteQueryText()))
+ return NOT_DELETE;
+ return solrDAO;
+ }
+
+ @Bean
+ @StepScope
public SolrDAO solrDAO(@Value("#{stepExecution.jobExecution.executionContext.get('jobProperties')}") DocumentExportProperties properties) {
return new SolrDAO(properties.getSolr());
}
diff --git a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentExportProperties.java b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentExportProperties.java
index 1484eed..d37edf4 100644
--- a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentExportProperties.java
+++ b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentExportProperties.java
@@ -31,13 +31,18 @@
import java.util.Optional;
import java.util.function.Supplier;
+import static java.util.Objects.requireNonNull;
+import static org.apache.ambari.infra.job.archive.ExportDestination.HDFS;
+import static org.apache.ambari.infra.job.archive.ExportDestination.LOCAL;
+import static org.apache.ambari.infra.job.archive.ExportDestination.S3;
import static org.apache.commons.csv.CSVFormat.DEFAULT;
import static org.apache.commons.lang.StringUtils.isBlank;
public class DocumentExportProperties extends JobProperties<DocumentExportProperties> {
private int readBlockSize;
private int writeBlockSize;
- private String destinationDirectoryPath;
+ private ExportDestination destination;
+ private String localDestinationDirectory;
private String fileNameSuffixColumn;
private String fileNameSuffixDateFormat;
private SolrProperties solr;
@@ -47,6 +52,9 @@
private String s3Endpoint;
private transient Supplier<Optional<S3Properties>> s3Properties;
+ private String hdfsEndpoint;
+ private String hdfsDestinationDirectory;
+
public DocumentExportProperties() {
super(DocumentExportProperties.class);
s3Properties = this::loadS3Properties;
@@ -101,12 +109,20 @@
this.writeBlockSize = writeBlockSize;
}
- public String getDestinationDirectoryPath() {
- return destinationDirectoryPath;
+ public ExportDestination getDestination() {
+ return destination;
}
- public void setDestinationDirectoryPath(String destinationDirectoryPath) {
- this.destinationDirectoryPath = destinationDirectoryPath;
+ public void setDestination(ExportDestination destination) {
+ this.destination = destination;
+ }
+
+ public String getLocalDestinationDirectory() {
+ return localDestinationDirectory;
+ }
+
+ public void setLocalDestinationDirectory(String localDestinationDirectory) {
+ this.localDestinationDirectory = localDestinationDirectory;
}
public String getFileNameSuffixColumn() {
@@ -170,11 +186,34 @@
return s3Properties.get();
}
+ public String getHdfsEndpoint() {
+ return hdfsEndpoint;
+ }
+
+ public void setHdfsEndpoint(String hdfsEndpoint) {
+ this.hdfsEndpoint = hdfsEndpoint;
+ }
+
+ public String getHdfsDestinationDirectory() {
+ return hdfsDestinationDirectory;
+ }
+
+ public void setHdfsDestinationDirectory(String hdfsDestinationDirectory) {
+ this.hdfsDestinationDirectory = hdfsDestinationDirectory;
+ }
+
@Override
public void apply(JobParameters jobParameters) {
readBlockSize = getIntJobParameter(jobParameters, "readBlockSize", readBlockSize);
writeBlockSize = getIntJobParameter(jobParameters, "writeBlockSize", writeBlockSize);
- destinationDirectoryPath = jobParameters.getString("destinationDirectoryPath", destinationDirectoryPath);
+ destination = ExportDestination.valueOf(jobParameters.getString("destination", destination.name()));
+ localDestinationDirectory = jobParameters.getString("localDestinationDirectory", localDestinationDirectory);
+ s3AccessFile = jobParameters.getString("s3AccessFile", s3AccessFile);
+ s3BucketName = jobParameters.getString("s3BucketName", s3BucketName);
+ s3KeyPrefix = jobParameters.getString("s3KeyPrefix", s3KeyPrefix);
+ s3Endpoint = jobParameters.getString("s3Endpoint", s3Endpoint);
+ hdfsEndpoint = jobParameters.getString("hdfsEndpoint", hdfsEndpoint);
+ hdfsDestinationDirectory = jobParameters.getString("hdfsDestinationDirectory", hdfsDestinationDirectory);
solr.apply(jobParameters);
}
@@ -193,13 +232,33 @@
if (writeBlockSize == 0)
throw new IllegalArgumentException("The property writeBlockSize must be greater than 0!");
- if (isBlank(destinationDirectoryPath))
- throw new IllegalArgumentException("The property destinationDirectoryPath can not be null or empty string!");
-
- if (isBlank(fileNameSuffixColumn))
+ if (isBlank(fileNameSuffixColumn)) {
throw new IllegalArgumentException("The property fileNameSuffixColumn can not be null or empty string!");
+ }
+
+ requireNonNull(destination, "The property destination can not be null!");
+ switch (destination) {
+ case LOCAL:
+ if (isBlank(localDestinationDirectory))
+ throw new IllegalArgumentException(String.format(
+ "The property localDestinationDirectory can not be null or empty string when destination is set to %s!", LOCAL.name()));
+ break;
+
+ case S3:
+ s3Properties()
+ .orElseThrow(() -> new IllegalArgumentException("S3 related properties must be set if the destination is " + S3.name()))
+ .validate();
+ break;
+
+ case HDFS:
+ if (isBlank(hdfsEndpoint))
+ throw new IllegalArgumentException(String.format(
+ "The property hdfsEndpoint can not be null or empty string when destination is set to %s!", HDFS.name()));
+ if (isBlank(hdfsDestinationDirectory))
+ throw new IllegalArgumentException(String.format(
+ "The property hdfsDestinationDirectory can not be null or empty string when destination is set to %s!", HDFS.name()));
+ }
solr.validate();
- s3Properties().ifPresent(S3Properties::validate);
}
}
diff --git a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/ExportDestination.java b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/ExportDestination.java
new file mode 100644
index 0000000..a143e4c
--- /dev/null
+++ b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/ExportDestination.java
@@ -0,0 +1,25 @@
+/*
+ * 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.ambari.infra.job.archive;
+
+public enum ExportDestination {
+ LOCAL,
+ HDFS,
+ S3
+}
diff --git a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/FileNameSuffixFormatter.java b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/FileNameSuffixFormatter.java
index af48ab9..9b4e662 100644
--- a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/FileNameSuffixFormatter.java
+++ b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/FileNameSuffixFormatter.java
@@ -21,6 +21,7 @@
import java.time.OffsetDateTime;
import java.time.format.DateTimeFormatter;
+import static java.util.Objects.requireNonNull;
import static org.apache.ambari.infra.job.archive.SolrDocumentIterator.SOLR_DATE_FORMAT_TEXT;
import static org.apache.commons.lang.StringUtils.isBlank;
@@ -42,15 +43,21 @@
}
public String format(Document document) {
- if (document == null)
- throw new NullPointerException("Can not format file name suffix: input document is null!");
+ requireNonNull(document, "Can not format file name suffix: input document is null!");
if (isBlank(document.get(columnName)))
throw new IllegalArgumentException("The specified document does not have a column " + columnName + " or it's value is blank!");
+ return format(document.get(columnName));
+ }
+
+ public String format(String value) {
+ if (isBlank(value))
+ throw new IllegalArgumentException("The specified value is blank!");
+
if (dateFormat == null)
- return document.get(columnName);
- OffsetDateTime date = OffsetDateTime.parse(document.get(columnName), SOLR_DATETIME_FORMATTER);
+ return value;
+ OffsetDateTime date = OffsetDateTime.parse(value, SOLR_DATETIME_FORMATTER);
return date.format(dateFormat);
}
}
diff --git a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/HdfsUploader.java b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/HdfsUploader.java
new file mode 100644
index 0000000..0f7b99f
--- /dev/null
+++ b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/HdfsUploader.java
@@ -0,0 +1,55 @@
+/*
+ * 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.ambari.infra.job.archive;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+
+public class HdfsUploader extends AbstractFileAction {
+
+ private final Configuration configuration;
+ private final Path destinationDirectory;
+
+ public HdfsUploader(Configuration configuration, Path destinationDirectory) {
+ this.destinationDirectory = destinationDirectory;
+ this.configuration = configuration;
+ }
+
+ @Override
+ protected File onPerform(File inputFile) {
+ try (FileSystem fileSystem = FileSystem.get(configuration)) {
+ Path destination = new Path(destinationDirectory, inputFile.getName());
+ if (fileSystem.exists(destination)) {
+ throw new UnsupportedOperationException(String.format("File '%s' already exists!", destination));
+ }
+
+ fileSystem.copyFromLocalFile(new Path(inputFile.getAbsolutePath()), destination);
+
+ return inputFile;
+ }
+ catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+}
diff --git a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/deleting/DocumentDeletingConfiguration.java b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/deleting/DocumentDeletingConfiguration.java
index 4fce4b6..ce8970d 100644
--- a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/deleting/DocumentDeletingConfiguration.java
+++ b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/deleting/DocumentDeletingConfiguration.java
@@ -58,6 +58,9 @@
@PostConstruct
public void createJobs() {
+ if (propertyMap == null || propertyMap.getSolrDataDeleting() == null)
+ return;
+
propertyMap.getSolrDataDeleting().values().forEach(DocumentDeletingProperties::validate);
propertyMap.getSolrDataDeleting().keySet().forEach(jobName -> {
diff --git a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/model/JobExecutionRestartRequest.java b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/model/JobExecutionRestartRequest.java
index 88687e7..3eab25f 100644
--- a/ambari-infra-manager/src/main/java/org/apache/ambari/infra/model/JobExecutionRestartRequest.java
+++ b/ambari-infra-manager/src/main/java/org/apache/ambari/infra/model/JobExecutionRestartRequest.java
@@ -18,12 +18,22 @@
*/
package org.apache.ambari.infra.model;
+import javax.validation.constraints.NotNull;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.QueryParam;
+
public class JobExecutionRestartRequest {
+ @PathParam("jobName")
+ @NotNull
private String jobName;
+ @PathParam("jobInstanceId")
+ @NotNull
private Long jobInstanceId;
+ @QueryParam("operation")
+ @NotNull
private JobOperationParams.JobRestartOperationParam operation;
public String getJobName() {
diff --git a/ambari-infra-manager/src/main/resources/infra-manager.properties b/ambari-infra-manager/src/main/resources/infra-manager.properties
index 9103d09..61c08e5 100644
--- a/ambari-infra-manager/src/main/resources/infra-manager.properties
+++ b/ambari-infra-manager/src/main/resources/infra-manager.properties
@@ -18,7 +18,7 @@
infra-manager.batch.db.password=admin
management.security.enabled=false
management.health.solr.enabled=false
-infra-manager.server.data.folder=/tmp
+infra-manager.server.data.folder=/tmp/ambariInfraManager
infra-manager.jobs.solr_data_export.export_service_logs.solr.zoo_keeper_connection_string=zookeeper:2181
infra-manager.jobs.solr_data_export.export_service_logs.solr.collection=hadoop_logs
@@ -28,9 +28,10 @@
infra-manager.jobs.solr_data_export.export_service_logs.solr.sort_column[1]=id
infra-manager.jobs.solr_data_export.export_service_logs.read_block_size=100
infra-manager.jobs.solr_data_export.export_service_logs.write_block_size=150
+infra-manager.jobs.solr_data_export.export_service_logs.destination=LOCAL
+infra-manager.jobs.solr_data_export.export_service_logs.local_destination_directory=/tmp/ambariInfraManager
infra-manager.jobs.solr_data_export.export_service_logs.file_name_suffix_column=logtime
infra-manager.jobs.solr_data_export.export_service_logs.file_name_suffix_date_format=yyyy-MM-dd'T'HH-mm-ss.SSSX
-infra-manager.jobs.solr_data_export.export_service_logs.destination_directory_path=/tmp/ambariInfraManager
infra-manager.jobs.solr_data_export.archive_audit_logs.solr.zoo_keeper_connection_string=zookeeper:2181
infra-manager.jobs.solr_data_export.archive_audit_logs.solr.collection=audit_logs
infra-manager.jobs.solr_data_export.archive_audit_logs.solr.query_text=logtime:[${start} TO ${end}]
@@ -40,10 +41,12 @@
infra-manager.jobs.solr_data_export.archive_audit_logs.solr.delete_query_text=logtime:[${start.logtime} TO ${end.logtime}} OR (logtime:${end.logtime} AND id:[* TO ${end.id}])
infra-manager.jobs.solr_data_export.archive_audit_logs.read_block_size=100
infra-manager.jobs.solr_data_export.archive_audit_logs.write_block_size=150
+infra-manager.jobs.solr_data_export.archive_audit_logs.destination=S3
# TODO: logtime may not be enough: The same filename can be generated when more than write_block_size count docs has the same logtime value
infra-manager.jobs.solr_data_export.archive_audit_logs.file_name_suffix_column=logtime
infra-manager.jobs.solr_data_export.archive_audit_logs.file_name_suffix_date_format=yyyy-MM-dd'T'HH-mm-ss.SSSX
-infra-manager.jobs.solr_data_export.archive_audit_logs.destination_directory_path=/tmp/ambariInfraManager
+infra-manager.jobs.solr_data_export.archive_audit_logs.hdfs_endpoint=hdfs://namenode:9000/
+infra-manager.jobs.solr_data_export.archive_audit_logs.hdfs_destination_directory=/test_audit_logs
#infra-manager.jobs.solr_data_export.archive_audit_logs.s3_access_file=<any>.csv
infra-manager.jobs.solr_data_export.archive_audit_logs.s3_key_prefix=solr_archive_
infra-manager.jobs.solr_data_export.archive_audit_logs.s3_bucket_name=testbucket
diff --git a/ambari-infra-manager/src/test/java/org/apache/ambari/infra/job/JobPropertiesTest.java b/ambari-infra-manager/src/test/java/org/apache/ambari/infra/job/JobPropertiesTest.java
index 6a76229..21bcdb7 100644
--- a/ambari-infra-manager/src/test/java/org/apache/ambari/infra/job/JobPropertiesTest.java
+++ b/ambari-infra-manager/src/test/java/org/apache/ambari/infra/job/JobPropertiesTest.java
@@ -29,7 +29,7 @@
@Test
public void testDeepCopy() throws Exception {
DocumentExportProperties documentExportProperties = new DocumentExportProperties();
- documentExportProperties.setDestinationDirectoryPath("/tmp");
+ documentExportProperties.setLocalDestinationDirectory("/tmp");
documentExportProperties.setFileNameSuffixColumn(".json");
documentExportProperties.setReadBlockSize(10);
documentExportProperties.setWriteBlockSize(20);
@@ -43,7 +43,7 @@
DocumentExportProperties parsed = documentExportProperties.deepCopy();
- assertThat(parsed.getDestinationDirectoryPath(), is(documentExportProperties.getDestinationDirectoryPath()));
+ assertThat(parsed.getLocalDestinationDirectory(), is(documentExportProperties.getLocalDestinationDirectory()));
assertThat(parsed.getFileNameSuffixColumn(), is(documentExportProperties.getFileNameSuffixColumn()));
assertThat(parsed.getReadBlockSize(), is(documentExportProperties.getReadBlockSize()));
assertThat(parsed.getWriteBlockSize(), is(documentExportProperties.getWriteBlockSize()));
diff --git a/ambari-infra-manager/src/test/java/org/apache/ambari/infra/job/archive/FileNameSuffixFormatterTest.java b/ambari-infra-manager/src/test/java/org/apache/ambari/infra/job/archive/FileNameSuffixFormatterTest.java
index 34e679f..cca2c1a 100644
--- a/ambari-infra-manager/src/test/java/org/apache/ambari/infra/job/archive/FileNameSuffixFormatterTest.java
+++ b/ambari-infra-manager/src/test/java/org/apache/ambari/infra/job/archive/FileNameSuffixFormatterTest.java
@@ -32,7 +32,7 @@
@Test(expected = NullPointerException.class)
public void testFormatWhenDocumentIsNullThrowingException() throws Exception {
- formatter.format(null);
+ formatter.format((Document) null);
}
@Test(expected = IllegalArgumentException.class)
diff --git a/ambari-infra-manager/src/test/resoruces/vagrant-infra-manager.properties.sample b/ambari-infra-manager/src/test/resoruces/vagrant-infra-manager.properties.sample
new file mode 100644
index 0000000..f008a53
--- /dev/null
+++ b/ambari-infra-manager/src/test/resoruces/vagrant-infra-manager.properties.sample
@@ -0,0 +1,37 @@
+# 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.
+infra-manager.batch.db.file=job-repository.db
+infra-manager.batch.db.init=true
+infra-manager.batch.db.username=admin
+infra-manager.batch.db.password=admin
+management.security.enabled=false
+management.health.solr.enabled=false
+infra-manager.server.data.folder=/tmp/ambariInfraManager
+
+infra-manager.jobs.solr_data_export.archive_service_logs.solr.zoo_keeper_connection_string=c6401.ambari.apache.org:2181/infra-solr
+infra-manager.jobs.solr_data_export.archive_service_logs.solr.collection=hadoop_logs
+infra-manager.jobs.solr_data_export.archive_service_logs.solr.query_text=logtime:[${start} TO ${end}]
+infra-manager.jobs.solr_data_export.archive_service_logs.solr.filter_query_text=(logtime:${logtime} AND id:{${id} TO *]) OR logtime:{${logtime} TO ${end}]
+infra-manager.jobs.solr_data_export.archive_service_logs.solr.sort_column[0]=logtime
+infra-manager.jobs.solr_data_export.archive_service_logs.solr.sort_column[1]=id
+infra-manager.jobs.solr_data_export.archive_service_logs.solr.delete_query_text=logtime:[${start.logtime} TO ${end.logtime}} OR (logtime:${end.logtime} AND id:[* TO ${end.id}])
+infra-manager.jobs.solr_data_export.archive_service_logs.read_block_size=2000
+infra-manager.jobs.solr_data_export.archive_service_logs.write_block_size=1000
+infra-manager.jobs.solr_data_export.archive_service_logs.destination=HDFS
+infra-manager.jobs.solr_data_export.archive_service_logs.file_name_suffix_column=logtime
+infra-manager.jobs.solr_data_export.archive_service_logs.file_name_suffix_date_format=yyyy-MM-dd'T'HH-mm-ss.SSSX
+infra-manager.jobs.solr_data_export.archive_service_logs.hdfs_endpoint=hdfs://c6401.ambari.apache.org:8020
+infra-manager.jobs.solr_data_export.archive_service_logs.hdfs_destination_directory=/archived_service_logs
+# Note: set hdfs user using the HADOOP_USER_NAME environmental variable. Value: hdfs
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 9e7a71b..2400493 100644
--- a/pom.xml
+++ b/pom.xml
@@ -36,6 +36,7 @@
<deb.python.ver>python (>= 2.6)</deb.python.ver>
<deb.architecture>amd64</deb.architecture>
<deb.dependency.list>${deb.python.ver}</deb.dependency.list>
+ <hadoop.version>3.0.0</hadoop.version>
</properties>
<modules>
