spaces vs tabs

README.md

@@ -204,6 +204,34 @@ By default, Beyonder will not overwrite a template if it already exists.
 This can be overridden by setting `force` to `true` in the expanded factory method
 `ElasticsearchBeyonder.start()`.
 
+Managing pipelines
+------------------
+
+A pipeline is a definition of a series of processors that are to be executed in the same order as they are declared while 
+documents are being indexed. Please note that this feature is only supported when you use the REST client not the Transport client.
+
+For example, setting one fields value based on another field by using an Set Processor you an add a file named `elasticsearch/_pipeline/set_field_processor`
+in your project:
+
+```javascript
+{
+  "description" : "Twitter pipeline",
+  "processors" : [
+    {
+      "set" : {
+        "field": "copy",
+        "value": "{{otherField}}"
+      }
+    }
+  ]
+}
+```
+
+By default, Beyonder will not overwrite a pipeline if it already exists.
+This can be overridden by setting `force` to `true` in the expanded factory method
+`ElasticsearchBeyonder.start()`.
+
+
 Why this name?
 ==============
 

pom.xml

@@ -22,12 +22,12 @@ when an elasticsearch client starts.
     </licenses>
 
     <properties>
-        <elasticsearch.version>7.7.1</elasticsearch.version>
+        <elasticsearch.version>7.8.1</elasticsearch.version>
         <test.elasticsearch.version>${elasticsearch.version}</test.elasticsearch.version>
         <slf4j.version>1.7.30</slf4j.version>
         <log4j.version>2.13.3</log4j.version>
         <commons.io.version>2.7</commons.io.version>
-        <jackson.version>2.11.0</jackson.version>
+        <jackson.version>2.11.2</jackson.version>
 
         <!-- For integration tests using Docker or external cluster -->
         <integ.elasticsearch.image>docker.elastic.co/elasticsearch/elasticsearch</integ.elasticsearch.image>
@@ -165,7 +165,7 @@ when an elasticsearch client starts.
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-text</artifactId>
-            <version>1.8</version>
+            <version>1.9</version>
         </dependency>
     </dependencies>
 
@@ -250,7 +250,7 @@ when an elasticsearch client starts.
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
-                <version>3.0.0-M4</version>
+                <version>3.0.0-M5</version>
                 <configuration>
                     <!-- Skips unit tests if the value of skipUnitTests property is true -->
                     <skipTests>${skipUnitTests}</skipTests>
@@ -266,7 +266,7 @@ when an elasticsearch client starts.
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-failsafe-plugin</artifactId>
-                <version>3.0.0-M4</version>
+                <version>3.0.0-M5</version>
                 <executions>
                     <!--
                         Invokes both the integration-test and the verify goals of the

src/main/java/fr/pilato/elasticsearch/tools/ElasticsearchBeyonder.java

@@ -22,6 +22,7 @@
 import fr.pilato.elasticsearch.tools.SettingsFinder.Defaults;
 import fr.pilato.elasticsearch.tools.index.IndexFinder;
 import fr.pilato.elasticsearch.tools.template.TemplateFinder;
+import fr.pilato.elasticsearch.tools.pipeline.PipelineFinder;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.client.RestClient;
 import org.slf4j.Logger;
@@ -33,6 +34,7 @@
 import static fr.pilato.elasticsearch.tools.index.IndexElasticsearchUpdater.createIndex;
 import static fr.pilato.elasticsearch.tools.index.IndexElasticsearchUpdater.updateSettings;
 import static fr.pilato.elasticsearch.tools.template.TemplateElasticsearchUpdater.createTemplate;
+import static fr.pilato.elasticsearch.tools.pipeline.PipelineElasticsearchUpdater.createPipeline;
 
 /**
  * By default, indexes are created with their default Elasticsearch settings. You can specify
@@ -109,6 +111,12 @@ public static void start(RestClient client, String root, boolean merge, boolean
 			createTemplate(client, root, templateName, force);
 		}
 
+		// create pipelines
+		List<String> pipelineNames = PipelineFinder.findPipelines(root);
+		for (String pipelineName : pipelineNames) {
+			createPipeline(client, root, pipelineName, force);
+		}
+
 		// create indices
 		Collection<String> indexNames = IndexFinder.findIndexNames(root);
 		for (String indexName : indexNames) {

src/main/java/fr/pilato/elasticsearch/tools/SettingsFinder.java

@@ -46,6 +46,7 @@ public static class Defaults {
 		public static String IndexSettingsFileName = "_settings.json";
 		public static String UpdateIndexSettingsFileName = "_update_settings.json";
 		public static String TemplateDir = "_template";
+		public static String PipelineDir = "_pipeline";
 
 		/**
 		 * Default setting of whether or not to merge mappings on start.

src/main/java/fr/pilato/elasticsearch/tools/index/IndexFinder.java

@@ -70,7 +70,7 @@ public static List<String> findIndexNames(final String root) throws IOException,
                 } else {
                     key = resource;
                 }
-                if (!key.equals(Defaults.TemplateDir) && !keys.contains(key)) {
+                if (!key.equals(Defaults.TemplateDir) && !key.equals(Defaults.PipelineDir) && !keys.contains(key)) {
                     logger.trace(" - found [{}].", key);
                     keys.add(key);
                     indexNames.add(key);

src/main/java/fr/pilato/elasticsearch/tools/pipeline/PipelineElasticsearchUpdater.java

@@ -0,0 +1,140 @@
+/*
+ * Licensed to David Pilato (the "Author") under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Author 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 fr.pilato.elasticsearch.tools.pipeline;
+
+import java.io.IOException;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.client.RestClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Handles ingest pipeline creation.
+ * 
+ * @author hjk181
+ *
+ */
+public class PipelineElasticsearchUpdater {
+
+    private static final Logger logger = LoggerFactory.getLogger(PipelineElasticsearchUpdater.class);
+
+    /**
+     * Create a pipeline in Elasticsearch.
+     * 
+     * @param client Elasticsearch client
+     * @param root dir within the classpath
+     * @param pipeline the id of the pipeline
+     * @param force set it to true if you want to force cleaning pipeline before adding it
+     * @throws Exception if something goes wrong
+     */
+    public static void createPipeline(RestClient client, String root, String pipeline, boolean force) throws Exception {
+        String json = PipelineSettingsReader.readPipeline(root, pipeline);
+        createPipelineWithJson(client, pipeline, json, force);
+    }
+
+    /**
+     * Create a pipeline in Elasticsearch. Read content from default classpath dir.
+     * 
+     * @param client Elasticsearch client
+     * @param pipeline the id of the pipeline
+     * @param force set it to true if you want to force cleaning pipeline before adding it
+     * @throws Exception if something goes wrong
+     */
+    public static void createPipeline(RestClient client, String pipeline, boolean force) throws Exception {
+        String json = PipelineSettingsReader.readPipeline(pipeline);
+        createPipelineWithJson(client, pipeline, json, force);
+    }
+
+    /**
+     * Create a new pipeline in Elasticsearch
+     * 
+     * @param client Elasticsearch client
+     * @param pipeline the id of the pipeline
+     * @param json JSon content for the pipeline
+     * @param force set it to true if you want to force cleaning pipeline before adding it
+     * @throws Exception if something goes wrong
+     */
+    public static void createPipelineWithJson(RestClient client, String pipeline, String json, boolean force) throws Exception {
+        if (isPipelineExist(client, pipeline)) {
+            if (force) {
+                logger.debug("Pipeline [{}] already exists. Force is set. Overriding it.", pipeline);
+                createPipelineWithJsonInElasticsearch(client, pipeline, json);
+            }
+            else {
+                logger.debug("Pipeline [{}] already exists.", pipeline);
+            }
+        }
+
+        if (!isPipelineExist(client, pipeline)) {
+            logger.debug("Pipeline [{}] doesn't exist. Creating it.", pipeline);
+            createPipelineWithJsonInElasticsearch(client, pipeline, json);
+        }
+    }
+
+    /**
+     * Create a new pipeline in Elasticsearch.
+     * 
+     * @param client Elasticsearch client
+     * @param pipeline the id of the pipeline
+     * @param json JSon content for the pipeline
+     * @throws Exception if something goes wrong
+     */
+    private static void createPipelineWithJsonInElasticsearch(RestClient client, String pipeline, String json) throws Exception {
+        logger.trace("createPipeline([{}])", pipeline);
+
+        assert client != null;
+        assert pipeline != null;
+
+        Request request = new Request("PUT", "/_ingest/pipeline/" + pipeline);
+        request.setJsonEntity(json);
+        Response response = client.performRequest(request);
+
+        if (response.getStatusLine().getStatusCode() != 200) {
+            logger.warn("Could not create pipeline [{}]", pipeline);
+            throw new Exception("Could not create pipeline [" + pipeline + "].");
+        }
+
+        logger.trace("/createPipeline([{}])", pipeline);
+    }
+
+    /**
+     * Check if a pipeline exists
+     * 
+     * @param client Elasticsearch client
+     * @param pipeline the id of the pipeline
+     * @return true if the pipeline exists
+     * @throws IOException if something goes wrong
+     */
+    public static boolean isPipelineExist(RestClient client, String pipeline) throws IOException {
+        try {
+            Response response = client.performRequest(new Request("GET", "/_ingest/pipeline/" + pipeline));
+            return response.getEntity() != null;
+        }
+        catch (ResponseException e) {
+            if (404 != e.getResponse().getStatusLine().getStatusCode()) {
+                throw e;
+            }
+        }
+        return false;
+    }
+}

src/main/java/fr/pilato/elasticsearch/tools/pipeline/PipelineFinder.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to David Pilato (the "Author") under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Author 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 fr.pilato.elasticsearch.tools.pipeline;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import fr.pilato.elasticsearch.tools.ResourceList;
+import fr.pilato.elasticsearch.tools.SettingsFinder;
+import fr.pilato.elasticsearch.tools.template.TemplateFinder;
+
+/**
+ * Findes ingest pipelines on the classpath.
+ * 
+ * @author hjk181
+ *
+ */
+public class PipelineFinder extends SettingsFinder {
+
+    private static final Logger logger = LoggerFactory.getLogger(TemplateFinder.class);
+
+    /**
+     * Find all pipelines in default classpath dir
+     * 
+     * @return a list of pipelines
+     * @throws IOException if connection with elasticsearch is failing
+     * @throws URISyntaxException this should not happen
+     */
+    public static List<String> findPipelines() throws IOException, URISyntaxException {
+        return findPipelines(Defaults.ConfigDir);
+    }
+
+    /**
+     * Find all pipelines
+     * 
+     * @param root dir within the classpath
+     * @return a list of pipelines
+     * @throws IOException if connection with elasticsearch is failing
+     * @throws URISyntaxException this should not happen
+     */
+    public static List<String> findPipelines(String root) throws IOException, URISyntaxException {
+        if (root == null) {
+            return findPipelines();
+        }
+
+        logger.debug("Looking for pipelines in classpath under [{}].", root);
+
+        final List<String> pipelineNames = new ArrayList<>();
+        String[] resources = ResourceList.getResources(root + "/" + Defaults.PipelineDir + "/"); // "es/_pipeline/"
+        for (String resource : resources) {
+            if (!resource.isEmpty()) {
+                String withoutIndex = resource.substring(resource.indexOf("/") + 1);
+                String pipeline = withoutIndex.substring(0, withoutIndex.indexOf(Defaults.JsonFileExtension));
+                logger.trace(" - found [{}].", pipeline);
+                pipelineNames.add(pipeline);
+            }
+        }
+
+        return pipelineNames;
+    }
+}