Watcher: Allow for external email attachments

This feature is mainly done for the integration with the commercial reporting, but can be used
for anything else as well.

This adds a `attachments` to the email configuration, which can be used like this

```
"attachments" : {
  "some_id" : {
    "http" : {
      "request" : {
        "url" : "http://example.org/foo.pdf"
      }
    }
  },
  "other_id" : {
    "data" : {
      "format" : "json"
    }
  }
}
```

The main reason to pick this format is extensibility. If we would like to support another
attachment type, like an file reader, we could do so easily from an API point of view.

Closes elastic/elasticsearch#870

Original commit: elastic/x-pack-elasticsearch@66d14be965

elasticsearch/qa/messy-test-xpack-with-mustache/src/test/java/org/elasticsearch/messy/tests/EmailAttachmentTests.java

@@ -0,0 +1,220 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.messy.tests;
+
+import com.squareup.okhttp.mockwebserver.MockResponse;
+import com.squareup.okhttp.mockwebserver.MockWebServer;
+import com.squareup.okhttp.mockwebserver.QueueDispatcher;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.Streams;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.script.mustache.MustachePlugin;
+import org.elasticsearch.watcher.actions.email.service.EmailTemplate;
+import org.elasticsearch.watcher.actions.email.service.attachment.DataAttachment;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentParser;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachments;
+import org.elasticsearch.watcher.actions.email.service.attachment.HttpRequestAttachment;
+import org.elasticsearch.watcher.actions.email.service.support.EmailServer;
+import org.elasticsearch.watcher.client.WatchSourceBuilder;
+import org.elasticsearch.watcher.client.WatcherClient;
+import org.elasticsearch.watcher.condition.compare.CompareCondition;
+import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
+import org.elasticsearch.watcher.support.http.Scheme;
+import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTestCase;
+import org.elasticsearch.watcher.trigger.schedule.IntervalSchedule;
+import org.junit.After;
+import org.junit.Before;
+
+import javax.mail.BodyPart;
+import javax.mail.Multipart;
+import javax.mail.Part;
+import javax.mail.internet.MimeMessage;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
+import static org.elasticsearch.search.builder.SearchSourceBuilder.searchSource;
+import static org.elasticsearch.watcher.actions.ActionBuilders.emailAction;
+import static org.elasticsearch.watcher.actions.email.DataAttachment.JSON;
+import static org.elasticsearch.watcher.actions.email.DataAttachment.YAML;
+import static org.elasticsearch.watcher.client.WatchSourceBuilders.watchBuilder;
+import static org.elasticsearch.watcher.condition.ConditionBuilders.compareCondition;
+import static org.elasticsearch.watcher.input.InputBuilders.searchInput;
+import static org.elasticsearch.watcher.test.WatcherTestUtils.newInputSearchRequest;
+import static org.elasticsearch.watcher.trigger.TriggerBuilders.schedule;
+import static org.elasticsearch.watcher.trigger.schedule.Schedules.interval;
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.startsWith;
+
+public class EmailAttachmentTests extends AbstractWatcherIntegrationTestCase {
+
+    static final String USERNAME = "_user";
+    static final String PASSWORD = "_passwd";
+
+    private MockWebServer webServer = new MockWebServer();;
+    private EmailServer server;
+
+    @Before
+    public void startWebservice() throws Exception {
+        QueueDispatcher dispatcher = new QueueDispatcher();
+        dispatcher.setFailFast(true);
+        webServer.setDispatcher(dispatcher);
+        webServer.start(0);
+        MockResponse mockResponse = new MockResponse().setResponseCode(200)
+                .addHeader("Content-Type", "application/foo").setBody("This is the content");
+        webServer.enqueue(mockResponse);
+    }
+
+    @After
+    public void cleanup() throws Exception {
+        server.stop();
+        webServer.shutdown();
+    }
+
+    @Override
+    protected List<Class<? extends Plugin>> pluginTypes() {
+        List<Class<? extends Plugin>> types = new ArrayList<>();
+        types.addAll(super.pluginTypes());
+        types.add(MustachePlugin.class);
+        return types;
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        if(server == null) {
+            //Need to construct the Email Server here as this happens before init()
+            server = EmailServer.localhost("2500-2600", USERNAME, PASSWORD, logger);
+        }
+        return Settings.builder()
+                .put(super.nodeSettings(nodeOrdinal))
+                .put("watcher.actions.email.service.account.test.smtp.auth", true)
+                .put("watcher.actions.email.service.account.test.smtp.user", USERNAME)
+                .put("watcher.actions.email.service.account.test.smtp.password", PASSWORD)
+                .put("watcher.actions.email.service.account.test.smtp.port", server.port())
+                .put("watcher.actions.email.service.account.test.smtp.host", "localhost")
+                .build();
+    }
+
+    public List<String> getAttachments(MimeMessage message) throws Exception {
+        Object content = message.getContent();
+        if (content instanceof String)
+            return null;
+
+        if (content instanceof Multipart) {
+            Multipart multipart = (Multipart) content;
+            List<String> result = new ArrayList<>();
+
+            for (int i = 0; i < multipart.getCount(); i++) {
+                result.addAll(getAttachments(multipart.getBodyPart(i)));
+            }
+            return result;
+
+        }
+        return null;
+    }
+
+    private List<String> getAttachments(BodyPart part) throws Exception {
+        List<String> result = new ArrayList<>();
+        Object content = part.getContent();
+        if (content instanceof InputStream || content instanceof String) {
+            if (Part.ATTACHMENT.equalsIgnoreCase(part.getDisposition()) || Strings.hasLength(part.getFileName())) {
+                result.add(Streams.copyToString(new InputStreamReader(part.getInputStream(), StandardCharsets.UTF_8)));
+                return result;
+            } else {
+                return new ArrayList<>();
+            }
+        }
+
+        if (content instanceof Multipart) {
+            Multipart multipart = (Multipart) content;
+            for (int i = 0; i < multipart.getCount(); i++) {
+                BodyPart bodyPart = multipart.getBodyPart(i);
+                result.addAll(getAttachments(bodyPart));
+            }
+        }
+        return result;
+    }
+
+    public void testThatEmailAttachmentsAreSent() throws Exception {
+        org.elasticsearch.watcher.actions.email.DataAttachment dataFormat = randomFrom(JSON, YAML);
+        final CountDownLatch latch = new CountDownLatch(1);
+        server.addListener(new EmailServer.Listener() {
+            @Override
+            public void on(MimeMessage message) throws Exception {
+                assertThat(message.getSubject(), equalTo("Subject"));
+                List<String> attachments = getAttachments(message);
+                if (dataFormat == YAML) {
+                    assertThat(attachments, hasItem(allOf(startsWith("---"), containsString("_test_id"))));
+                } else {
+                    assertThat(attachments, hasItem(allOf(startsWith("{"), containsString("_test_id"))));
+                }
+                assertThat(attachments, hasItem(containsString("This is the content")));
+                latch.countDown();
+            }
+        });
+
+        WatcherClient watcherClient = watcherClient();
+        createIndex("idx");
+        // Have a sample document in the index, the watch is going to evaluate
+        client().prepareIndex("idx", "type").setSource("field", "value").get();
+        refresh();
+        SearchRequest searchRequest = newInputSearchRequest("idx").source(searchSource().query(matchAllQuery()));
+
+        List<EmailAttachmentParser.EmailAttachment> attachments = new ArrayList<>();
+
+        DataAttachment dataAttachment = DataAttachment.builder("my-id").dataAttachment(dataFormat).build();
+        attachments.add(dataAttachment);
+
+        HttpRequestTemplate requestTemplate = HttpRequestTemplate.builder("localhost", webServer.getPort()).path("/").scheme(Scheme.HTTP).build();
+        HttpRequestAttachment httpRequestAttachment = HttpRequestAttachment.builder("other-id").httpRequestTemplate(requestTemplate).build();
+
+        attachments.add(httpRequestAttachment);
+        EmailAttachments emailAttachments = new EmailAttachments(attachments);
+        XContentBuilder tmpBuilder = jsonBuilder();
+        emailAttachments.toXContent(tmpBuilder, ToXContent.EMPTY_PARAMS);
+        logger.info("TMP BUILDER {}", tmpBuilder.string());
+
+        EmailTemplate.Builder emailBuilder = EmailTemplate.builder().from("_from").to("_to").subject("Subject");
+        WatchSourceBuilder watchSourceBuilder = watchBuilder()
+                .trigger(schedule(interval(5, IntervalSchedule.Interval.Unit.SECONDS)))
+                .input(searchInput(searchRequest))
+                .condition(compareCondition("ctx.payload.hits.total", CompareCondition.Op.GT, 0l))
+                .addAction("_email", emailAction(emailBuilder).setAuthentication(USERNAME, PASSWORD.toCharArray())
+                .setAttachments(emailAttachments));
+        logger.info("TMP WATCHSOURCE {}", watchSourceBuilder.build().getBytes().toUtf8());
+
+        watcherClient.preparePutWatch("_test_id")
+                .setSource(watchSourceBuilder)
+                .get();
+
+        if (timeWarped()) {
+            timeWarp().scheduler().trigger("_test_id");
+            refresh();
+        }
+
+        assertWatchWithMinimumPerformedActionsCount("_test_id", 1);
+
+        if (!latch.await(5, TimeUnit.SECONDS)) {
+            fail("waited too long for email to be received");
+        }
+    }
+
+
+
+}

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/WatcherActionModule.java

@@ -12,6 +12,10 @@ import org.elasticsearch.watcher.actions.email.EmailActionFactory;
 import org.elasticsearch.watcher.actions.email.service.EmailService;
 import org.elasticsearch.watcher.actions.email.service.HtmlSanitizer;
 import org.elasticsearch.watcher.actions.email.service.InternalEmailService;
+import org.elasticsearch.watcher.actions.email.service.attachment.DataAttachmentParser;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentParser;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentsParser;
+import org.elasticsearch.watcher.actions.email.service.attachment.HttpEmailAttachementParser;
 import org.elasticsearch.watcher.actions.hipchat.HipChatAction;
 import org.elasticsearch.watcher.actions.hipchat.HipChatActionFactory;
 import org.elasticsearch.watcher.actions.hipchat.service.HipChatService;
@@ -39,6 +43,7 @@ import java.util.Map;
 public class WatcherActionModule extends AbstractModule {
 
     private final Map<String, Class<? extends ActionFactory>> parsers = new HashMap<>();
+    private final Map<String, Class<? extends EmailAttachmentParser>> emailAttachmentParsers = new HashMap<>();
 
     public WatcherActionModule() {
         registerAction(EmailAction.TYPE, EmailActionFactory.class);
@@ -48,12 +53,19 @@ public class WatcherActionModule extends AbstractModule {
         registerAction(HipChatAction.TYPE, HipChatActionFactory.class);
         registerAction(SlackAction.TYPE, SlackActionFactory.class);
         registerAction(PagerDutyAction.TYPE, PagerDutyActionFactory.class);
+
+        registerEmailAttachmentParser(HttpEmailAttachementParser.TYPE, HttpEmailAttachementParser.class);
+        registerEmailAttachmentParser(DataAttachmentParser.TYPE, DataAttachmentParser.class);
     }
 
     public void registerAction(String type, Class<? extends ActionFactory> parserType) {
         parsers.put(type, parserType);
     }
 
+    public void registerEmailAttachmentParser(String type, Class<? extends EmailAttachmentParser> parserClass) {
+        emailAttachmentParsers.put(type, parserClass);
+    }
+
     @Override
     protected void configure() {
 
@@ -70,6 +82,12 @@ public class WatcherActionModule extends AbstractModule {
         bind(InternalEmailService.class).asEagerSingleton();
         bind(EmailService.class).to(InternalEmailService.class).asEagerSingleton();
 
+        MapBinder<String, EmailAttachmentParser> emailParsersBinder = MapBinder.newMapBinder(binder(), String.class, EmailAttachmentParser.class);
+        for (Map.Entry<String, Class<? extends EmailAttachmentParser>> entry : emailAttachmentParsers.entrySet()) {
+            emailParsersBinder.addBinding(entry.getKey()).to(entry.getValue()).asEagerSingleton();
+        }
+        bind(EmailAttachmentsParser.class).asEagerSingleton();
+
         // hipchat
         bind(InternalHipChatService.class).asEagerSingleton();
         bind(HipChatService.class).to(InternalHipChatService.class);

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/EmailAction.java

@@ -16,6 +16,8 @@ import org.elasticsearch.watcher.actions.email.service.Authentication;
 import org.elasticsearch.watcher.actions.email.service.Email;
 import org.elasticsearch.watcher.actions.email.service.EmailTemplate;
 import org.elasticsearch.watcher.actions.email.service.Profile;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachments;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentsParser;
 import org.elasticsearch.watcher.support.secret.Secret;
 import org.elasticsearch.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.watcher.support.xcontent.WatcherXContentParser;
@@ -35,13 +37,15 @@ public class EmailAction implements Action {
     private final @Nullable Authentication auth;
     private final @Nullable Profile profile;
     private final @Nullable DataAttachment dataAttachment;
+    private final @Nullable EmailAttachments emailAttachments;
 
-    public EmailAction(EmailTemplate email, @Nullable String account, @Nullable Authentication auth, @Nullable Profile profile, @Nullable DataAttachment dataAttachment) {
+    public EmailAction(EmailTemplate email, @Nullable String account, @Nullable Authentication auth, @Nullable Profile profile, @Nullable DataAttachment dataAttachment, @Nullable EmailAttachments emailAttachments) {
         this.email = email;
         this.account = account;
         this.auth = auth;
         this.profile = profile;
         this.dataAttachment = dataAttachment;
+        this.emailAttachments = emailAttachments;
     }
 
     public EmailTemplate getEmail() {
@@ -64,6 +68,10 @@ public class EmailAction implements Action {
         return dataAttachment;
     }
 
+    public EmailAttachments getAttachments() {
+        return emailAttachments;
+    }
+
     @Override
     public String type() {
         return TYPE;
@@ -80,6 +88,7 @@ public class EmailAction implements Action {
         if (account != null ? !account.equals(action.account) : action.account != null) return false;
         if (auth != null ? !auth.equals(action.auth) : action.auth != null) return false;
         if (profile != action.profile) return false;
+        if (emailAttachments != null && !emailAttachments.equals(action.emailAttachments)) return false;
         return !(dataAttachment != null ? !dataAttachment.equals(action.dataAttachment) : action.dataAttachment != null);
     }
 
@@ -90,6 +99,7 @@ public class EmailAction implements Action {
         result = 31 * result + (auth != null ? auth.hashCode() : 0);
         result = 31 * result + (profile != null ? profile.hashCode() : 0);
         result = 31 * result + (dataAttachment != null ? dataAttachment.hashCode() : 0);
+        result = 31 * result + (emailAttachments != null ? emailAttachments.hashCode() : 0);
         return result;
     }
 
@@ -111,17 +121,21 @@ public class EmailAction implements Action {
         if (dataAttachment != null) {
             builder.field(Field.ATTACH_DATA.getPreferredName(), dataAttachment, params);
         }
+        if (emailAttachments != null) {
+            emailAttachments.toXContent(builder, params);
+        }
         email.xContentBody(builder, params);
         return builder.endObject();
     }
 
-    public static EmailAction parse(String watchId, String actionId, XContentParser parser) throws IOException {
+    public static EmailAction parse(String watchId, String actionId, XContentParser parser, EmailAttachmentsParser emailAttachmentsParser) throws IOException {
         EmailTemplate.Parser emailParser = new EmailTemplate.Parser();
         String account = null;
         String user = null;
         Secret password = null;
         Profile profile = Profile.STANDARD;
         DataAttachment dataAttachment = null;
+        EmailAttachments attachments = null;
 
         String currentFieldName = null;
         XContentParser.Token token;
@@ -134,6 +148,8 @@ public class EmailAction implements Action {
                 } catch (IOException ioe) {
                     throw new ElasticsearchParseException("could not parse [{}] action [{}/{}]. failed to parse data attachment field [{}]", ioe, TYPE, watchId, actionId, currentFieldName);
                 }
+            } else if (ParseFieldMatcher.STRICT.match(currentFieldName, Field.ATTACHMENTS)) {
+                attachments = emailAttachmentsParser.parse(parser);
             } else if (!emailParser.handle(currentFieldName, parser)) {
                 if (token == XContentParser.Token.VALUE_STRING) {
                     if (ParseFieldMatcher.STRICT.match(currentFieldName, Field.ACCOUNT)) {
@@ -162,7 +178,7 @@ public class EmailAction implements Action {
             auth = new Authentication(user, password);
         }
 
-        return new EmailAction(emailParser.parsedTemplate(), account, auth, profile, dataAttachment);
+        return new EmailAction(emailParser.parsedTemplate(), account, auth, profile, dataAttachment, attachments);
     }
 
     public static Builder builder(EmailTemplate email) {
@@ -232,6 +248,7 @@ public class EmailAction implements Action {
         @Nullable Authentication auth;
         @Nullable Profile profile;
         @Nullable DataAttachment dataAttachment;
+        @Nullable EmailAttachments attachments;
 
         private Builder(EmailTemplate email) {
             this.email = email;
@@ -252,13 +269,19 @@ public class EmailAction implements Action {
             return this;
         }
 
+        @Deprecated
         public Builder setAttachPayload(DataAttachment dataAttachment) {
             this.dataAttachment = dataAttachment;
             return this;
         }
 
+        public Builder setAttachments(EmailAttachments attachments) {
+            this.attachments = attachments;
+            return this;
+        }
+
         public EmailAction build() {
-            return new EmailAction(email, account, auth, profile, dataAttachment);
+            return new EmailAction(email, account, auth, profile, dataAttachment, attachments);
         }
     }
 
@@ -272,6 +295,7 @@ public class EmailAction implements Action {
         ParseField USER = new ParseField("user");
         ParseField PASSWORD = new ParseField("password");
         ParseField ATTACH_DATA = new ParseField("attach_data");
+        ParseField ATTACHMENTS = new ParseField("attachments");
 
         // result fields
         ParseField MESSAGE = new ParseField("message");

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/EmailActionFactory.java

@@ -12,9 +12,12 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.watcher.actions.ActionFactory;
 import org.elasticsearch.watcher.actions.email.service.EmailService;
 import org.elasticsearch.watcher.actions.email.service.HtmlSanitizer;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentParser;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentsParser;
 import org.elasticsearch.watcher.support.text.TextTemplateEngine;
 
 import java.io.IOException;
+import java.util.Map;
 
 /**
  *
@@ -24,13 +27,18 @@ public class EmailActionFactory extends ActionFactory<EmailAction, ExecutableEma
     private final EmailService emailService;
     private final TextTemplateEngine templateEngine;
     private final HtmlSanitizer htmlSanitizer;
+    private final EmailAttachmentsParser emailAttachmentsParser;
+    private final Map<String, EmailAttachmentParser> emailAttachmentParsers;
 
     @Inject
-    public EmailActionFactory(Settings settings, EmailService emailService, TextTemplateEngine templateEngine, HtmlSanitizer htmlSanitizer) {
+    public EmailActionFactory(Settings settings, EmailService emailService, TextTemplateEngine templateEngine, HtmlSanitizer htmlSanitizer,
+                              EmailAttachmentsParser emailAttachmentsParser, Map<String, EmailAttachmentParser> emailAttachmentParsers) {
         super(Loggers.getLogger(ExecutableEmailAction.class, settings));
         this.emailService = emailService;
         this.templateEngine = templateEngine;
         this.htmlSanitizer = htmlSanitizer;
+        this.emailAttachmentsParser = emailAttachmentsParser;
+        this.emailAttachmentParsers = emailAttachmentParsers;
     }
 
     @Override
@@ -40,11 +48,11 @@ public class EmailActionFactory extends ActionFactory<EmailAction, ExecutableEma
 
     @Override
     public EmailAction parseAction(String watchId, String actionId, XContentParser parser) throws IOException {
-        return EmailAction.parse(watchId, actionId, parser);
+        return EmailAction.parse(watchId, actionId, parser, emailAttachmentsParser);
     }
 
     @Override
     public ExecutableEmailAction createExecutable(EmailAction action) {
-        return new ExecutableEmailAction(action, actionLogger, emailService, templateEngine, htmlSanitizer);
+        return new ExecutableEmailAction(action, actionLogger, emailService, templateEngine, htmlSanitizer, emailAttachmentParsers);
     }
 }

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/ExecutableEmailAction.java

@@ -12,6 +12,7 @@ import org.elasticsearch.watcher.actions.email.service.Attachment;
 import org.elasticsearch.watcher.actions.email.service.Email;
 import org.elasticsearch.watcher.actions.email.service.EmailService;
 import org.elasticsearch.watcher.actions.email.service.HtmlSanitizer;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentParser;
 import org.elasticsearch.watcher.execution.WatchExecutionContext;
 import org.elasticsearch.watcher.support.Variables;
 import org.elasticsearch.watcher.support.text.TextTemplateEngine;
@@ -27,12 +28,15 @@ public class ExecutableEmailAction extends ExecutableAction<EmailAction> {
     final EmailService emailService;
     final TextTemplateEngine templateEngine;
     final HtmlSanitizer htmlSanitizer;
+    private final Map<String, EmailAttachmentParser> emailAttachmentParsers;
 
-    public ExecutableEmailAction(EmailAction action, ESLogger logger, EmailService emailService, TextTemplateEngine templateEngine, HtmlSanitizer htmlSanitizer) {
+    public ExecutableEmailAction(EmailAction action, ESLogger logger, EmailService emailService, TextTemplateEngine templateEngine, HtmlSanitizer htmlSanitizer,
+                                 Map<String, EmailAttachmentParser> emailAttachmentParsers) {
         super(action, logger);
         this.emailService = emailService;
         this.templateEngine = templateEngine;
         this.htmlSanitizer = htmlSanitizer;
+        this.emailAttachmentParsers = emailAttachmentParsers;
     }
 
     public Action.Result execute(String actionId, WatchExecutionContext ctx, Payload payload) throws Exception {
@@ -45,6 +49,14 @@ public class ExecutableEmailAction extends ExecutableAction<EmailAction> {
             attachments.put(attachment.id(), attachment);
         }
 
+        if (action.getAttachments() != null && action.getAttachments().getAttachments().size() > 0) {
+            for (EmailAttachmentParser.EmailAttachment emailAttachment : action.getAttachments().getAttachments()) {
+                EmailAttachmentParser parser = emailAttachmentParsers.get(emailAttachment.type());
+                Attachment attachment = parser.toAttachment(ctx, payload, emailAttachment);
+                attachments.put(attachment.id(), attachment);
+            }
+        }
+
         Email.Builder email = action.getEmail().render(templateEngine, model, htmlSanitizer, attachments);
         email.id(ctx.id().value());
 

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/service/attachment/DataAttachment.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+
+public class DataAttachment implements EmailAttachmentParser.EmailAttachment {
+
+    private final String id;
+    private final org.elasticsearch.watcher.actions.email.DataAttachment dataAttachment;
+
+    public DataAttachment(String id, org.elasticsearch.watcher.actions.email.DataAttachment dataAttachment) {
+        this.id = id;
+        this.dataAttachment = dataAttachment;
+    }
+
+    public org.elasticsearch.watcher.actions.email.DataAttachment getDataAttachment() {
+        return dataAttachment;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject(id).startObject(DataAttachmentParser.TYPE);
+        if (dataAttachment == org.elasticsearch.watcher.actions.email.DataAttachment.YAML) {
+            builder.field("format", "yaml");
+        } else {
+            builder.field("format", "json");
+        }
+
+        return builder.endObject().endObject();
+    }
+
+    @Override
+    public String type() {
+        return DataAttachmentParser.TYPE;
+    }
+
+    public static Builder builder(String id) {
+        return new Builder(id);
+    }
+
+
+    public static class Builder {
+
+        private String id;
+        private org.elasticsearch.watcher.actions.email.DataAttachment dataAttachment;
+
+        private Builder(String id) {
+            this.id = id;
+        }
+
+        public Builder dataAttachment(org.elasticsearch.watcher.actions.email.DataAttachment dataAttachment) {
+            this.dataAttachment = dataAttachment;
+            return this;
+        }
+
+        public DataAttachment build() {
+            return new DataAttachment(id, dataAttachment);
+        }
+    }
+}

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/service/attachment/DataAttachmentParser.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.ParseFieldMatcher;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.watcher.actions.email.service.Attachment;
+import org.elasticsearch.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.watcher.support.Variables;
+import org.elasticsearch.watcher.watch.Payload;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.elasticsearch.watcher.actions.email.DataAttachment.resolve;
+
+public class DataAttachmentParser implements EmailAttachmentParser<DataAttachment> {
+
+    interface Fields {
+        ParseField FORMAT = new ParseField("format");
+    }
+
+    public static final String TYPE = "data";
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public DataAttachment parse(String id, XContentParser parser) throws IOException {
+        org.elasticsearch.watcher.actions.email.DataAttachment dataAttachment = org.elasticsearch.watcher.actions.email.DataAttachment.YAML;
+
+        String currentFieldName = null;
+        XContentParser.Token token;
+        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == XContentParser.Token.FIELD_NAME) {
+                currentFieldName = parser.currentName();
+            } else if (Strings.hasLength(currentFieldName) && ParseFieldMatcher.STRICT.match(currentFieldName, Fields.FORMAT)) {
+                if (token == XContentParser.Token.VALUE_STRING) {
+                    dataAttachment = resolve(parser.text());
+                } else {
+                    throw new ElasticsearchParseException("could not parse data attachment. expected string value for [{}] field but found [{}] instead", currentFieldName, token);
+                }
+            }
+        }
+
+        return new DataAttachment(id, dataAttachment);
+    }
+
+    @Override
+    public Attachment toAttachment(WatchExecutionContext ctx, Payload payload, DataAttachment attachment) {
+        Map<String, Object> model = Variables.createCtxModel(ctx, payload);
+        return attachment.getDataAttachment().create(model);
+    }
+}

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/service/attachment/EmailAttachmentParser.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.watcher.actions.email.service.Attachment;
+import org.elasticsearch.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.watcher.watch.Payload;
+
+import java.io.IOException;
+
+/**
+ * Marker interface for email attachments that have an additional execution step and are used by
+ * EmailAttachmentParser class
+ */
+public interface EmailAttachmentParser<T extends EmailAttachmentParser.EmailAttachment> {
+
+    interface EmailAttachment extends ToXContent {
+        /**
+         * @return A type to identify the email attachment, same as the parser identifier
+         */
+        String type();
+    }
+
+    /**
+     * @return An identifier of this parser
+     */
+    String type();
+
+    /**
+     * A parser to create an EmailAttachment, that is serializable and does not execute anything
+     *
+     * @param id The id of this attachment, parsed from the outer content
+     * @param parser The XContentParser used for parsing
+     * @return A concrete EmailAttachment
+     * @throws IOException in case parsing fails
+     */
+    T parse(String id, XContentParser parser) throws IOException;
+
+    /**
+     * Converts an email attachment to an attachment, potentially executing code like an HTTP request
+     * @param context The WatchExecutionContext supplied with the whole watch execution
+     * @param payload The Payload supplied with the action
+     * @param attachment The typed attachment
+     * @return An attachment that is ready to be used in a MimeMessage
+     */
+    Attachment toAttachment(WatchExecutionContext context, Payload payload, T attachment);
+
+}

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/service/attachment/EmailAttachments.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.List;
+
+public class EmailAttachments implements ToXContent {
+
+    public interface Fields {
+        ParseField ATTACHMENTS = new ParseField("attachments");
+    }
+
+    private final List<EmailAttachmentParser.EmailAttachment> attachments;
+
+    public EmailAttachments(List<EmailAttachmentParser.EmailAttachment> attachments) {
+        this.attachments = attachments;
+    }
+
+    public List<EmailAttachmentParser.EmailAttachment> getAttachments() {
+        return attachments;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        if (attachments != null && attachments.size() > 0) {
+            builder.startObject(Fields.ATTACHMENTS.getPreferredName());
+            for (EmailAttachmentParser.EmailAttachment attachment : attachments) {
+                attachment.toXContent(builder, params);
+            }
+            builder.endObject();
+        }
+
+        return builder;
+    }
+}

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/service/attachment/EmailAttachmentsParser.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class EmailAttachmentsParser {
+
+    private Map<String, EmailAttachmentParser> parsers;
+
+    @Inject
+    public EmailAttachmentsParser(Map<String, EmailAttachmentParser> parsers) {
+        this.parsers = parsers;
+    }
+
+    public EmailAttachments parse(XContentParser parser) throws IOException {
+        List<EmailAttachmentParser.EmailAttachment> attachments = new ArrayList<>();
+        String currentFieldName = null;
+        XContentParser.Token token;
+        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == XContentParser.Token.FIELD_NAME) {
+                currentFieldName = parser.currentName();
+            } else {
+                if (token == XContentParser.Token.START_OBJECT && currentFieldName != null) {
+                    String currentAttachmentType = null;
+                    if (parser.nextToken() == XContentParser.Token.FIELD_NAME) {
+                        currentAttachmentType = parser.currentName();
+                    }
+                    parser.nextToken();
+
+                    EmailAttachmentParser emailAttachmentParser = parsers.get(currentAttachmentType);
+                    if (emailAttachmentParser == null) {
+                        throw new ElasticsearchParseException("Cannot parse attachment of type " + currentAttachmentType);
+                    }
+                    EmailAttachmentParser.EmailAttachment emailAttachment = emailAttachmentParser.parse(currentFieldName, parser);
+                    attachments.add(emailAttachment);
+                    // one further to skip the end_object from the attachment
+                    parser.nextToken();
+                }
+            }
+        }
+
+        return new EmailAttachments(attachments);
+    }
+
+}

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/service/attachment/HttpEmailAttachementParser.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.ParseFieldMatcher;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.watcher.actions.email.service.Attachment;
+import org.elasticsearch.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.watcher.support.Variables;
+import org.elasticsearch.watcher.support.http.HttpClient;
+import org.elasticsearch.watcher.support.http.HttpRequest;
+import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
+import org.elasticsearch.watcher.support.http.HttpResponse;
+import org.elasticsearch.watcher.support.text.TextTemplateEngine;
+import org.elasticsearch.watcher.watch.Payload;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class HttpEmailAttachementParser implements EmailAttachmentParser<HttpRequestAttachment> {
+
+    public interface Fields {
+        ParseField REQUEST = new ParseField("request");
+        ParseField CONTENT_TYPE = new ParseField("content_type");
+    }
+
+    public static final String TYPE = "http";
+    private final HttpClient httpClient;
+    private HttpRequestTemplate.Parser requestTemplateParser;
+    private final TextTemplateEngine templateEngine;
+    private final ESLogger logger;
+
+    @Inject
+    public HttpEmailAttachementParser(HttpClient httpClient, HttpRequestTemplate.Parser requestTemplateParser, TextTemplateEngine templateEngine) {
+        this.httpClient = httpClient;
+        this.requestTemplateParser = requestTemplateParser;
+        this.templateEngine = templateEngine;
+        this.logger = Loggers.getLogger(getClass());
+    }
+
+    @Override
+    public String type() {
+        return TYPE;
+    }
+
+    @Override
+    public HttpRequestAttachment parse(String id, XContentParser parser) throws IOException {
+        String contentType = null;
+        HttpRequestTemplate requestTemplate = null;
+
+        String currentFieldName = null;
+        XContentParser.Token token;
+        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == XContentParser.Token.FIELD_NAME) {
+                currentFieldName = parser.currentName();
+            } else if (ParseFieldMatcher.STRICT.match(currentFieldName, Fields.CONTENT_TYPE)) {
+                contentType = parser.text();
+            } else if (ParseFieldMatcher.STRICT.match(currentFieldName, Fields.REQUEST)) {
+                requestTemplate = requestTemplateParser.parse(parser);
+            } else {
+                throw new ElasticsearchParseException("Unknown field name [" + currentFieldName + "] in http request attachment configuration");
+            }
+        }
+
+        if (requestTemplate != null) {
+            return new HttpRequestAttachment(id, requestTemplate, contentType);
+        }
+
+        throw new ElasticsearchParseException("Could not parse http request attachment");
+    }
+
+    @Override
+    public Attachment toAttachment(WatchExecutionContext context, Payload payload, HttpRequestAttachment attachment) {
+        Map<String, Object> model = Variables.createCtxModel(context, payload);
+        HttpRequest httpRequest = attachment.getRequestTemplate().render(templateEngine, model);
+
+        try {
+            HttpResponse response = httpClient.execute(httpRequest);
+            // check for status 200, only then append attachment
+            if (response.status() >= 200 && response.status() < 300) {
+                if (response.hasContent()) {
+                    String contentType = attachment.getContentType();
+                    String attachmentContentType = Strings.hasLength(contentType) ? contentType : response.contentType();
+                    return new Attachment.Bytes(attachment.getId(), response.body().toBytes(), attachmentContentType);
+                } else {
+                    logger.error("Empty response body: [host[{}], port[{}], method[{}], path[{}]: response status [{}]", httpRequest.host(),
+                            httpRequest.port(), httpRequest.method(), httpRequest.path(), response.status());
+                }
+            } else {
+                logger.error("Error getting http response: [host[{}], port[{}], method[{}], path[{}]: response status [{}]", httpRequest.host(),
+                        httpRequest.port(), httpRequest.method(), httpRequest.path(), response.status());
+            }
+        } catch (IOException e) {
+            logger.error("Error executing HTTP request: [host[{}], port[{}], method[{}], path[{}]: [{}]", e, httpRequest.port(),
+                    httpRequest.method(), httpRequest.path(), e.getMessage());
+        }
+
+        return null;
+    }
+}

elasticsearch/x-pack/watcher/src/main/java/org/elasticsearch/watcher/actions/email/service/attachment/HttpRequestAttachment.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
+
+import java.io.IOException;
+
+public class HttpRequestAttachment implements EmailAttachmentParser.EmailAttachment {
+
+    private final HttpRequestTemplate requestTemplate;
+    private final String contentType;
+    private String id;
+
+    public HttpRequestAttachment(String id, HttpRequestTemplate requestTemplate, @Nullable String contentType) {
+        this.id = id;
+        this.requestTemplate = requestTemplate;
+        this.contentType = contentType;
+    }
+
+    public HttpRequestTemplate getRequestTemplate() {
+        return requestTemplate;
+    }
+
+    public String getContentType() {
+        return contentType;
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject(id)
+                .startObject(HttpEmailAttachementParser.TYPE)
+                .field(HttpEmailAttachementParser.Fields.REQUEST.getPreferredName(), requestTemplate, params);
+        if (Strings.hasLength(contentType)) {
+            builder.field(HttpEmailAttachementParser.Fields.CONTENT_TYPE.getPreferredName(), contentType);
+        }
+        return builder.endObject().endObject();
+    }
+
+    public static Builder builder(String id) {
+        return new Builder(id);
+    }
+
+    @Override
+    public String type() {
+        return HttpEmailAttachementParser.TYPE;
+    }
+
+    public static class Builder {
+
+        private String id;
+        private HttpRequestTemplate httpRequestTemplate;
+        private String contentType;
+
+        private Builder(String id) {
+            this.id = id;
+        }
+
+        public Builder httpRequestTemplate(HttpRequestTemplate httpRequestTemplate) {
+            this.httpRequestTemplate = httpRequestTemplate;
+            return this;
+        }
+
+        public Builder contentType(String contentType) {
+            this.contentType = contentType;
+            return this;
+        }
+
+        public HttpRequestAttachment build() {
+            return new HttpRequestAttachment(id, httpRequestTemplate, contentType);
+        }
+
+    }
+}

elasticsearch/x-pack/watcher/src/test/java/org/elasticsearch/watcher/actions/email/EmailActionTests.java

@@ -5,9 +5,11 @@
  */
 package org.elasticsearch.watcher.actions.email;
 
+import com.google.common.base.Charsets;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.MapBuilder;
+import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -15,23 +17,40 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.watcher.actions.Action;
+import org.elasticsearch.watcher.actions.email.service.Attachment;
 import org.elasticsearch.watcher.actions.email.service.Authentication;
 import org.elasticsearch.watcher.actions.email.service.Email;
 import org.elasticsearch.watcher.actions.email.service.EmailService;
 import org.elasticsearch.watcher.actions.email.service.EmailTemplate;
 import org.elasticsearch.watcher.actions.email.service.HtmlSanitizer;
 import org.elasticsearch.watcher.actions.email.service.Profile;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentParser;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachments;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentsParser;
+import org.elasticsearch.watcher.actions.email.service.attachment.HttpEmailAttachementParser;
 import org.elasticsearch.watcher.execution.WatchExecutionContext;
 import org.elasticsearch.watcher.execution.Wid;
+import org.elasticsearch.watcher.support.http.HttpClient;
+import org.elasticsearch.watcher.support.http.HttpRequest;
+import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
+import org.elasticsearch.watcher.support.http.HttpResponse;
+import org.elasticsearch.watcher.support.http.auth.HttpAuthRegistry;
+import org.elasticsearch.watcher.support.http.auth.basic.BasicAuthFactory;
 import org.elasticsearch.watcher.support.secret.Secret;
+import org.elasticsearch.watcher.support.secret.SecretService;
 import org.elasticsearch.watcher.support.text.TextTemplate;
 import org.elasticsearch.watcher.support.text.TextTemplateEngine;
 import org.elasticsearch.watcher.support.xcontent.WatcherParams;
 import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTestCase;
 import org.elasticsearch.watcher.watch.Payload;
+import org.jboss.netty.handler.codec.http.HttpHeaders;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -43,10 +62,12 @@ import static org.hamcrest.Matchers.arrayContainingInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -54,6 +75,12 @@ import static org.mockito.Mockito.when;
  *
  */
 public class EmailActionTests extends ESTestCase {
+
+    private SecretService secretService = mock(SecretService.class);
+    private HttpAuthRegistry registry = new HttpAuthRegistry(singletonMap("basic", new BasicAuthFactory(secretService)));
+    private HttpClient httpClient = mock(HttpClient.class);
+    private static final EmailAttachmentsParser EMPTY_EMAIL_ATTACHMENTS_PARSER = new EmailAttachmentsParser(Collections.emptyMap());
+
     public void testExecute() throws Exception {
         final String account = "account1";
         EmailService service = new AbstractWatcherIntegrationTestCase.NoopEmailService() {
@@ -92,9 +119,11 @@ public class EmailActionTests extends ESTestCase {
         Profile profile = randomFrom(Profile.values());
 
         DataAttachment dataAttachment = randomDataAttachment();
+        // TODO RANDOMIZE ME
+        EmailAttachments emailAttachments = new EmailAttachments(new ArrayList<>());
 
-        EmailAction action = new EmailAction(email, account, auth, profile, dataAttachment);
+        EmailAction action = new EmailAction(email, account, auth, profile, dataAttachment, emailAttachments);
-        ExecutableEmailAction executable = new ExecutableEmailAction(action, logger, service, engine, htmlSanitizer);
+        ExecutableEmailAction executable = new ExecutableEmailAction(action, logger, service, engine, htmlSanitizer, Collections.emptyMap());
 
         Map<String, Object> data = new HashMap<>();
         Payload payload = new Payload.Simple(data);
@@ -253,7 +282,7 @@ public class EmailActionTests extends ESTestCase {
         XContentParser parser = JsonXContent.jsonXContent.createParser(bytes);
         parser.nextToken();
 
-        ExecutableEmailAction executable = new EmailActionFactory(Settings.EMPTY, emailService, engine, htmlSanitizer)
+        ExecutableEmailAction executable = new EmailActionFactory(Settings.EMPTY, emailService, engine, htmlSanitizer, EMPTY_EMAIL_ATTACHMENTS_PARSER, Collections.emptyMap())
                 .parseExecutable(randomAsciiOfLength(8), randomAsciiOfLength(3), parser);
 
         assertThat(executable, notNullValue());
@@ -331,9 +360,11 @@ public class EmailActionTests extends ESTestCase {
         Profile profile = randomFrom(Profile.values());
         String account = randomAsciiOfLength(6);
         DataAttachment dataAttachment = randomDataAttachment();
+        // TODO randomize
+        EmailAttachments emailAttachments = new EmailAttachments(new ArrayList<>());
 
-        EmailAction action = new EmailAction(email, account, auth, profile, dataAttachment);
+        EmailAction action = new EmailAction(email, account, auth, profile, dataAttachment, emailAttachments);
-        ExecutableEmailAction executable = new ExecutableEmailAction(action, logger, service, engine, htmlSanitizer);
+        ExecutableEmailAction executable = new ExecutableEmailAction(action, logger, service, engine, htmlSanitizer, Collections.emptyMap());
 
         boolean hideSecrets = randomBoolean();
         ToXContent.Params params = WatcherParams.builder().hideSecrets(hideSecrets).build();
@@ -344,7 +375,8 @@ public class EmailActionTests extends ESTestCase {
         logger.info(bytes.toUtf8());
         XContentParser parser = JsonXContent.jsonXContent.createParser(bytes);
         parser.nextToken();
-        ExecutableEmailAction parsed = new EmailActionFactory(Settings.EMPTY, service, engine, htmlSanitizer)
+
+        ExecutableEmailAction parsed = new EmailActionFactory(Settings.EMPTY, service, engine, htmlSanitizer, EMPTY_EMAIL_ATTACHMENTS_PARSER, Collections.emptyMap())
                 .parseExecutable(randomAsciiOfLength(4), randomAsciiOfLength(10), parser);
 
         if (!hideSecrets) {
@@ -369,18 +401,100 @@ public class EmailActionTests extends ESTestCase {
         EmailService emailService = mock(EmailService.class);
         TextTemplateEngine engine = mock(TextTemplateEngine.class);
         HtmlSanitizer htmlSanitizer = mock(HtmlSanitizer.class);
+        EmailAttachmentsParser emailAttachmentsParser = mock(EmailAttachmentsParser.class);
+
         XContentBuilder builder = jsonBuilder().startObject().field("unknown_field", "value");
         XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
         parser.nextToken();
         try {
-            new EmailActionFactory(Settings.EMPTY, emailService, engine, htmlSanitizer)
+            new EmailActionFactory(Settings.EMPTY, emailService, engine, htmlSanitizer, emailAttachmentsParser, Collections.emptyMap())
                     .parseExecutable(randomAsciiOfLength(3), randomAsciiOfLength(7), parser);
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), containsString("unexpected string field [unknown_field]"));
         }
     }
 
+    public void testRequestAttachmentGetsAppendedToEmailAttachments() throws Exception {
+        String attachmentId = "my_attachment";
+
+        EmailService emailService = new AbstractWatcherIntegrationTestCase.NoopEmailService();
+        TextTemplateEngine engine = mock(TextTemplateEngine.class);
+        HtmlSanitizer htmlSanitizer = mock(HtmlSanitizer.class);
+        HttpClient httpClient = mock(HttpClient.class);
+
+        // setup mock response
+        Map<String, String[]> headers = new HashMap<>(1);
+        headers.put(HttpHeaders.Names.CONTENT_TYPE, new String[]{"plain/text"});
+        String content = "My wonderful text";
+        HttpResponse mockResponse = new HttpResponse(200, content, headers);
+        when(httpClient.execute(any(HttpRequest.class))).thenReturn(mockResponse);
+
+        // setup email parsers
+        HttpRequestTemplate.Parser httpRequestTemplateParser = new HttpRequestTemplate.Parser(registry);
+        Map<String, EmailAttachmentParser> attachmentParsers = new HashMap<>();
+        attachmentParsers.put(HttpEmailAttachementParser.TYPE, new HttpEmailAttachementParser(httpClient, httpRequestTemplateParser, engine));
+        EmailAttachmentsParser emailAttachmentsParser = new EmailAttachmentsParser(attachmentParsers);
+
+        XContentBuilder builder = jsonBuilder().startObject()
+                .startObject("attachments")
+                .startObject(attachmentId)
+                .startObject("http")
+                .startObject("request")
+                .field("host", "localhost")
+                .field("port", 443)
+                .field("path", "/the/evil/test")
+                .endObject()
+                .endObject()
+                .endObject()
+                .endObject()
+                .endObject();
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        logger.info("JSON: {}", builder.string());
+
+        parser.nextToken();
+
+        ExecutableEmailAction executableEmailAction = new EmailActionFactory(Settings.EMPTY, emailService, engine, htmlSanitizer, emailAttachmentsParser, attachmentParsers)
+                .parseExecutable(randomAsciiOfLength(3), randomAsciiOfLength(7), parser);
+
+        DateTime now = DateTime.now(DateTimeZone.UTC);
+        Wid wid = new Wid(randomAsciiOfLength(5), randomLong(), now);
+        Map<String, Object> metadata = MapBuilder.<String, Object>newMapBuilder().put("_key", "_val").map();
+        WatchExecutionContext ctx = mockExecutionContextBuilder("watch1")
+                .wid(wid)
+                .payload(new Payload.Simple())
+                .time("watch1", now)
+                .metadata(metadata)
+                .buildMock();
+
+        Action.Result result = executableEmailAction.execute("test", ctx, new Payload.Simple());
+        assertThat(result, instanceOf(EmailAction.Result.Success.class));
+
+        EmailAction.Result.Success successResult = (EmailAction.Result.Success) result;
+        Map<String, Attachment> attachments = successResult.email().attachments();
+        assertThat(attachments.keySet(), hasSize(1));
+        assertThat(attachments, hasKey(attachmentId));
+        Attachment externalAttachment = attachments.get(attachmentId);
+
+        assertThat(externalAttachment.bodyPart(), is(notNullValue()));
+        InputStream is = externalAttachment.bodyPart().getInputStream();
+        String data = Streams.copyToString(new InputStreamReader(is, Charsets.UTF_8));
+        assertThat(data, is(content));
+    }
+
     static DataAttachment randomDataAttachment() {
         return randomFrom(DataAttachment.JSON, DataAttachment.YAML, null);
     }
+
+    private HttpRequest randomExternalAttachment() throws Exception {
+        if (randomBoolean()) {
+            Map<String, String[]> headers = new HashMap<>(1);
+            headers.put(HttpHeaders.Names.CONTENT_TYPE, new String[]{"plain/text"});
+            String content = "My wonderful text";
+            HttpResponse mockResponse = new HttpResponse(200, content, headers);
+            when(httpClient.execute(any(HttpRequest.class))).thenReturn(mockResponse);
+            return HttpRequest.builder("_host", 443).build();
+        } else {
+            return null;
+        }
+    }
 }

elasticsearch/x-pack/watcher/src/test/java/org/elasticsearch/watcher/actions/email/service/attachment/DataAttachmentParserTests.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.core.Is.is;
+
+public class DataAttachmentParserTests extends ESTestCase {
+
+    public void testSerializationWorks() throws Exception {
+        Map<String, EmailAttachmentParser> attachmentParsers = new HashMap<>();
+        attachmentParsers.put(DataAttachmentParser.TYPE, new DataAttachmentParser());
+        EmailAttachmentsParser emailAttachmentsParser = new EmailAttachmentsParser(attachmentParsers);
+
+        String id = "some-id";
+        XContentBuilder builder = jsonBuilder().startObject().startObject(id)
+                .startObject(DataAttachmentParser.TYPE).field("format", randomFrom("yaml", "json")).endObject()
+                .endObject().endObject();
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        logger.info("JSON: {}", builder.string());
+
+        EmailAttachments emailAttachments = emailAttachmentsParser.parse(parser);
+        assertThat(emailAttachments.getAttachments(), hasSize(1));
+
+        XContentBuilder toXcontentBuilder = jsonBuilder().startObject();
+        emailAttachments.getAttachments().get(0).toXContent(toXcontentBuilder, ToXContent.EMPTY_PARAMS);
+        toXcontentBuilder.endObject();
+        assertThat(toXcontentBuilder.string(), is(builder.string()));
+    }
+
+}

elasticsearch/x-pack/watcher/src/test/java/org/elasticsearch/watcher/actions/email/service/attachment/EmailAttachmentParsersTests.java

@@ -0,0 +1,180 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import com.google.common.base.Charsets;
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.watcher.actions.email.service.Attachment;
+import org.elasticsearch.watcher.execution.WatchExecutionContext;
+import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
+import org.elasticsearch.watcher.support.http.Scheme;
+import org.elasticsearch.watcher.watch.Payload;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.core.Is.is;
+import static org.mockito.Mockito.mock;
+
+public class EmailAttachmentParsersTests extends ESTestCase {
+
+    private WatchExecutionContext ctx = mock(WatchExecutionContext.class);
+
+    public void testThatCustomParsersCanBeRegistered() throws Exception {
+        Map<String, EmailAttachmentParser> parsers = new HashMap<>();
+        parsers.put("test", new TestEmailAttachmentParser());
+        EmailAttachmentsParser parser = new EmailAttachmentsParser(parsers);
+
+        XContentBuilder builder = jsonBuilder();
+        builder.startObject()
+                .startObject("my-id")
+                .startObject("test")
+                .field("foo", "bar")
+                .endObject()
+                .endObject()
+                .startObject("my-other-id")
+                .startObject("test")
+                .field("foo", "baz")
+                .endObject()
+                .endObject()
+                .endObject();
+
+        logger.info("JSON: {}", builder.string());
+        XContentParser xContentParser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        EmailAttachments attachments = parser.parse(xContentParser);
+        assertThat(attachments.getAttachments(), hasSize(2));
+
+        EmailAttachmentParser.EmailAttachment emailAttachment = attachments.getAttachments().get(0);
+        assertThat(emailAttachment, instanceOf(TestEmailAttachment.class));
+
+        Attachment attachment = parsers.get("test").toAttachment(ctx, new Payload.Simple(), emailAttachment);
+        assertThat(attachment.name(), is("my-id"));
+        assertThat(attachment.contentType(), is("personalContentType"));
+
+        assertThat(parsers.get("test").toAttachment(ctx, new Payload.Simple(), attachments.getAttachments().get(1)).id(), is("my-other-id"));
+    }
+
+    public void testThatUnknownParserThrowsException() throws IOException {
+        EmailAttachmentsParser parser = new EmailAttachmentsParser(Collections.emptyMap());
+
+        XContentBuilder builder = jsonBuilder();
+        String type = randomAsciiOfLength(8);
+        builder.startObject().startObject("some-id").startObject(type);
+
+        XContentParser xContentParser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        try {
+            parser.parse(xContentParser);
+            fail("Expected random parser of type [" + type + "] to throw an exception");
+        } catch (ElasticsearchParseException e) {
+            assertThat(e.getMessage(), containsString("Cannot parse attachment of type " + type));
+        }
+    }
+
+    public void testThatToXContentSerializationWorks() throws Exception {
+        List<EmailAttachmentParser.EmailAttachment> attachments = new ArrayList<>();
+        attachments.add(new DataAttachment("my-id", org.elasticsearch.watcher.actions.email.DataAttachment.JSON));
+
+        HttpRequestTemplate requestTemplate = HttpRequestTemplate.builder("localhost", 80).scheme(Scheme.HTTP).path("/").build();
+        HttpRequestAttachment httpRequestAttachment = new HttpRequestAttachment("other-id", requestTemplate, null);
+
+        attachments.add(httpRequestAttachment);
+        EmailAttachments emailAttachments = new EmailAttachments(attachments);
+        XContentBuilder builder = jsonBuilder();
+        emailAttachments.toXContent(builder, ToXContent.EMPTY_PARAMS);
+        logger.info("JSON is: " + builder.string());
+        assertThat(builder.string(), containsString("my-id"));
+        assertThat(builder.string(), containsString("json"));
+        assertThat(builder.string(), containsString("other-id"));
+        assertThat(builder.string(), containsString("localhost"));
+        assertThat(builder.string(), containsString("/"));
+    }
+
+    public class TestEmailAttachmentParser implements EmailAttachmentParser<TestEmailAttachment> {
+
+        @Override
+        public String type() {
+            return "test";
+        }
+
+        @Override
+        public TestEmailAttachment parse(String id, XContentParser parser) throws IOException {
+            TestEmailAttachment attachment = null;
+            String currentFieldName = null;
+            XContentParser.Token token;
+            while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+                if (token == XContentParser.Token.FIELD_NAME) {
+                    currentFieldName = parser.currentName();
+                } else {
+                    if ("foo".equals(currentFieldName)) {
+                        attachment = new TestEmailAttachment(id, parser.text());
+                    }
+                }
+            }
+
+            if (attachment == null) {
+                throw new ElasticsearchParseException("Expected test parser to have field [foo]");
+            }
+
+            return attachment;
+        }
+
+        @Override
+        public Attachment toAttachment(WatchExecutionContext ctx, Payload payload, TestEmailAttachment attachment) {
+            return new Attachment.Bytes(attachment.getId(), attachment.getValue().getBytes(Charsets.UTF_8), "personalContentType");
+        }
+    }
+
+    public static class TestEmailAttachment implements EmailAttachmentParser.EmailAttachment {
+
+        private final String value;
+        private final String id;
+
+        interface Fields {
+            ParseField FOO = new ParseField("foo");
+        }
+
+        public TestEmailAttachment(String id, String value) {
+            this.id = id;
+            this.value = value;
+        }
+
+        @Override
+        public String type() {
+            return "test";
+        }
+
+        public String getValue() {
+            return value;
+        }
+
+        public String getId() {
+            return id;
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            return builder.startObject(id)
+                    .startObject(type())
+                    .field(Fields.FOO.getPreferredName(), value)
+                    .endObject()
+                    .endObject();
+        }
+    }
+}

elasticsearch/x-pack/watcher/src/test/java/org/elasticsearch/watcher/actions/email/service/attachment/HttpEmailAttachementParserTests.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.actions.email.service.attachment;
+
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.watcher.support.http.HttpClient;
+import org.elasticsearch.watcher.support.http.HttpRequest;
+import org.elasticsearch.watcher.support.http.HttpRequestTemplate;
+import org.elasticsearch.watcher.support.http.HttpRequestTemplateTests;
+import org.elasticsearch.watcher.support.http.HttpResponse;
+import org.elasticsearch.watcher.support.http.auth.HttpAuthRegistry;
+import org.elasticsearch.watcher.support.http.auth.basic.BasicAuth;
+import org.elasticsearch.watcher.support.http.auth.basic.BasicAuthFactory;
+import org.elasticsearch.watcher.support.secret.SecretService;
+import org.junit.Before;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Collections.singletonMap;
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.core.Is.is;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class HttpEmailAttachementParserTests extends ESTestCase {
+
+    private SecretService.PlainText secretService;
+    private HttpAuthRegistry authRegistry;
+    private HttpRequestTemplate.Parser httpRequestTemplateParser;
+    private HttpClient httpClient;
+
+    @Before
+    public void init() throws Exception {
+        secretService = new SecretService.PlainText();
+        authRegistry = new HttpAuthRegistry(singletonMap(BasicAuth.TYPE, new BasicAuthFactory(secretService)));
+        httpRequestTemplateParser = new HttpRequestTemplate.Parser(authRegistry);
+        httpClient = mock(HttpClient.class);
+
+        HttpResponse response = new HttpResponse(200, "This is my response".getBytes(UTF_8));
+        when(httpClient.execute(any(HttpRequest.class))).thenReturn(response);
+    }
+
+
+    public void testSerializationWorks() throws Exception {
+        Map<String, EmailAttachmentParser> attachmentParsers = new HashMap<>();
+        attachmentParsers.put(HttpEmailAttachementParser.TYPE, new HttpEmailAttachementParser(httpClient, httpRequestTemplateParser, new HttpRequestTemplateTests.MockTextTemplateEngine()));
+        EmailAttachmentsParser emailAttachmentsParser = new EmailAttachmentsParser(attachmentParsers);
+
+        String id = "some-id";
+        XContentBuilder builder = jsonBuilder().startObject().startObject(id)
+                .startObject(HttpEmailAttachementParser.TYPE)
+                .startObject("request")
+                .field("scheme", "http")
+                .field("host", "test.de")
+                .field("port", 80)
+                .field("method", "get")
+                .field("path", "/foo")
+                .startObject("params").endObject()
+                .startObject("headers").endObject()
+                .endObject();
+
+        boolean configureContentType = randomBoolean();
+        if (configureContentType) {
+            builder.field("content_type", "application/foo");
+        }
+        builder.endObject().endObject().endObject();
+        XContentParser parser = JsonXContent.jsonXContent.createParser(builder.bytes());
+        logger.info("JSON: {}", builder.string());
+
+        EmailAttachments emailAttachments = emailAttachmentsParser.parse(parser);
+        assertThat(emailAttachments.getAttachments(), hasSize(1));
+
+        XContentBuilder toXcontentBuilder = jsonBuilder().startObject();
+        emailAttachments.getAttachments().get(0).toXContent(toXcontentBuilder, ToXContent.EMPTY_PARAMS);
+        toXcontentBuilder.endObject();
+        assertThat(toXcontentBuilder.string(), is(builder.string()));
+    }
+
+}

elasticsearch/x-pack/watcher/src/test/java/org/elasticsearch/watcher/support/http/HttpRequestTemplateTests.java

@@ -169,7 +169,7 @@ public class HttpRequestTemplateTests extends ESTestCase {
         assertThat(parsedTemplate, is(urlParsedTemplate));
     }
 
-    static class MockTextTemplateEngine implements TextTemplateEngine {
+    public static class MockTextTemplateEngine implements TextTemplateEngine {
         @Override
         public String render(TextTemplate template, Map<String, Object> model) {
             return template.getTemplate();

elasticsearch/x-pack/watcher/src/test/java/org/elasticsearch/watcher/test/WatcherTestUtils.java

@@ -74,6 +74,7 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
@@ -210,8 +211,8 @@ public final class WatcherTestUtils {
 
         Authentication auth = new Authentication("testname", new Secret("testpassword".toCharArray()));
 
-        EmailAction action = new EmailAction(email, "testaccount", auth, Profile.STANDARD, null);
+        EmailAction action = new EmailAction(email, "testaccount", auth, Profile.STANDARD, null, null);
-        ExecutableEmailAction executale = new ExecutableEmailAction(action, logger, emailService, templateEngine, new HtmlSanitizer(Settings.EMPTY));
+        ExecutableEmailAction executale = new ExecutableEmailAction(action, logger, emailService, templateEngine, new HtmlSanitizer(Settings.EMPTY), Collections.emptyMap());
 
         actions.add(new ActionWrapper("_email", executale));
 

elasticsearch/x-pack/watcher/src/test/java/org/elasticsearch/watcher/watch/WatchTests.java

@@ -30,6 +30,7 @@ import org.elasticsearch.watcher.actions.email.service.EmailService;
 import org.elasticsearch.watcher.actions.email.service.EmailTemplate;
 import org.elasticsearch.watcher.actions.email.service.HtmlSanitizer;
 import org.elasticsearch.watcher.actions.email.service.Profile;
+import org.elasticsearch.watcher.actions.email.service.attachment.EmailAttachmentsParser;
 import org.elasticsearch.watcher.actions.index.ExecutableIndexAction;
 import org.elasticsearch.watcher.actions.index.IndexAction;
 import org.elasticsearch.watcher.actions.index.IndexActionFactory;
@@ -420,8 +421,8 @@ public class WatchTests extends ESTestCase {
         List<ActionWrapper> list = new ArrayList<>();
         if (randomBoolean()) {
             ExecutableTransform transform = randomTransform();
-            EmailAction action = new EmailAction(EmailTemplate.builder().build(), null, null, Profile.STANDARD, randomFrom(DataAttachment.JSON, DataAttachment.YAML, null));
+            EmailAction action = new EmailAction(EmailTemplate.builder().build(), null, null, Profile.STANDARD, randomFrom(DataAttachment.JSON, DataAttachment.YAML, null), null);
-            list.add(new ActionWrapper("_email_" + randomAsciiOfLength(8), randomThrottler(), transform, new ExecutableEmailAction(action, logger, emailService, templateEngine, htmlSanitizer)));
+            list.add(new ActionWrapper("_email_" + randomAsciiOfLength(8), randomThrottler(), transform, new ExecutableEmailAction(action, logger, emailService, templateEngine, htmlSanitizer, Collections.emptyMap())));
         }
         if (randomBoolean()) {
             DateTimeZone timeZone = randomBoolean() ? DateTimeZone.UTC : null;
@@ -445,7 +446,7 @@ public class WatchTests extends ESTestCase {
         for (ActionWrapper action : actions) {
             switch (action.action().type()) {
                 case EmailAction.TYPE:
-                    parsers.put(EmailAction.TYPE, new EmailActionFactory(settings, emailService, templateEngine, htmlSanitizer));
+                    parsers.put(EmailAction.TYPE, new EmailActionFactory(settings, emailService, templateEngine, htmlSanitizer, new EmailAttachmentsParser(Collections.emptyMap()), Collections.emptyMap()));
                     break;
                 case IndexAction.TYPE:
                     parsers.put(IndexAction.TYPE, new IndexActionFactory(settings, client));