FLUME-3415 - Provide Docker image for Flume

.github/workflows/build.yml

@@ -65,14 +65,34 @@ jobs:
       - name: Show disk usage before maven
         run: df -kh
 
-      - name: Build with Maven
-        timeout-minutes: 120
+      - name: Build with Maven (Linux)
+        if: runner.os == 'Linux'
+        timeout-minutes: 90
         shell: bash
         run: |
           ./mvnw clean verify -DredirectTestOutput=true \
           --show-version --batch-mode --errors --no-transfer-progress \
           -DtrimStackTrace=false \
           -Dsurefire.rerunFailingTestsCount=2
+
+      - name: Deploy Docker (Linux)
+        if: runner.os == 'Linux' && github.ref == 'refs/heads/trunk'
+        timeout-minutes: 90
+        shell: bash
+        run: |
+          ./mvnw deploy -Ddockerfile.username=$${{ secrets.DOCKERHUB_USER }} \
+           -Ddockerfile.password=... $${{ secrets.DOCKERHUB_TOKEN }}
+
+      - name: Build with Maven (MacOS)
+        if: runner.os == 'macOS'
+        timeout-minutes: 120
+        shell: bash
+        run: |
+          ./mvnw clean verify -DredirectTestOutput=true \
+          --show-version --batch-mode --errors --no-transfer-progress \
+          -DtrimStackTrace=false \
+          -Dsurefire.rerunFailingTestsCount=2 \
+          -P skipDocker
 
       - name: Show disk usage after
-        run: df -kh        
+        run: df -kh        

conf/log4j2.xml

@@ -50,6 +50,7 @@
     <Logger name="org.apache.hadoop" level="INFO"/>
     <Logger name="org.apache.hadoop.hive" level="ERROR"/>
     <Root level="INFO">
+      <AppenderRef ref="Console" />
       <AppenderRef ref="LogFile" />
     </Root>
   </Loggers>

flume-ng-dist/Dockerfile

@@ -0,0 +1,39 @@
+# 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.
+#
+
+ARG IMAGE_NAME=adoptopenjdk/openjdk8
+ARG IMAGE_TAG=centos-jre
+
+FROM ${IMAGE_NAME}:${IMAGE_TAG}
+
+ARG ASSEMBLY_VERSION
+ARG MAINTAINER="Apache Flume <[email protected]>"
+
+LABEL org.opencontainers.image.authors="${MAINTAINER}"
+LABEL site="https://flume.apache.org"
+
+COPY target/apache-flume-${ASSEMBLY_VERSION}-bin/apache-flume-${ASSEMBLY_VERSION}-bin/ /etc/flume/
+
+RUN echo $ASSEMBLY_VERSION
+
+RUN groupadd --system flume && useradd --system --shell /bin/false -g flume flume
+
+WORKDIR /etc/flume
+RUN chown -R flume:flume /etc/flume
+USER flume
+ENTRYPOINT ["/etc/flume/bin/flume-ng"]

flume-ng-dist/pom.xml

@@ -91,6 +91,29 @@
           <skip>true</skip>
         </configuration>
       </plugin>
+      <plugin>
+        <groupId>com.spotify</groupId>
+        <artifactId>dockerfile-maven-plugin</artifactId>
+        <version>${dockerfile.version}</version>
+        <!-- Wire up to the default build phases -->
+        <executions>
+          <execution>
+            <id>default</id>
+            <goals>
+              <goal>build</goal>
+              <goal>push</goal>
+            </goals>
+          </execution>
+        </executions>
+        <configuration>
+          <repository>apache/flume</repository>
+          <tag>${project.version}</tag>
+          <buildArgs>
+            <ASSEMBLY_VERSION>${project.version}</ASSEMBLY_VERSION>
+          </buildArgs>
+          <skip>${docker.skip}</skip>
+        </configuration>
+      </plugin>
     </plugins>
   </build>
 

flume-ng-tests/src/test/java/org/apache/flume/test/agent/TestFileChannelDocker.java

@@ -0,0 +1,155 @@
+/*
+ * 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.flume.test.agent;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import org.apache.flume.test.util.DockerInstall;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.ArrayList;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+public class TestFileChannelDocker {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(TestFileChannelDocker.class);
+
+  private static final List<File> tempResources = new ArrayList<File>();
+
+  private Properties agentProps;
+  private File sinkOutputDir;
+
+  @Before
+  public void setUp() throws Exception {
+    /* Create 3 temp dirs, each used as value within agentProps */
+
+    final File sinkOutputDir = Files.createTempDir();
+    sinkOutputDir.setWritable(true, false);
+    sinkOutputDir.setReadable(true, false);
+    tempResources.add(sinkOutputDir.getCanonicalFile());
+    final String sinkOutputDirPath = sinkOutputDir.getCanonicalPath();
+    LOGGER.info("Created rolling file sink's output dir: " + sinkOutputDirPath);
+
+    final File channelCheckpointDir = Files.createTempDir();
+    channelCheckpointDir.setWritable(true, false);
+    channelCheckpointDir.setReadable(true, false);
+    tempResources.add(channelCheckpointDir.getCanonicalFile());
+    final String channelCheckpointDirPath = channelCheckpointDir.getCanonicalPath();
+    LOGGER.info("Created file channel's checkpoint dir: " + channelCheckpointDirPath);
+
+    final File channelDataDir = Files.createTempDir();
+    channelDataDir.setWritable(true, false);
+    channelDataDir.setReadable(true, false);
+    tempResources.add(channelDataDir.getCanonicalFile());
+    final String channelDataDirPath = channelDataDir.getCanonicalPath();
+    LOGGER.info("Created file channel's data dir: " + channelDataDirPath);
+
+    /* Build props to pass to flume agent */
+
+    Properties agentProps = new Properties();
+
+    // Active sets
+    agentProps.put("a1.channels", "c1");
+    agentProps.put("a1.sources", "r1");
+    agentProps.put("a1.sinks", "k1");
+
+    // c1
+    agentProps.put("a1.channels.c1.type", "FILE");
+    agentProps.put("a1.channels.c1.checkpointDir", channelCheckpointDirPath);
+    agentProps.put("a1.channels.c1.dataDirs", channelDataDirPath);
+
+    // r1
+    agentProps.put("a1.sources.r1.channels", "c1");
+    agentProps.put("a1.sources.r1.type", "EXEC");
+    agentProps.put("a1.sources.r1.command", "seq 1 100");
+
+    // k1
+    agentProps.put("a1.sinks.k1.channel", "c1");
+    agentProps.put("a1.sinks.k1.type", "FILE_ROLL");
+    agentProps.put("a1.sinks.k1.sink.directory", sinkOutputDirPath);
+    agentProps.put("a1.sinks.k1.sink.rollInterval", "0");
+
+    this.agentProps = agentProps;
+    this.sinkOutputDir = sinkOutputDir;
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    DockerInstall.getInstance().stopAgent();
+    for (File tempResource : tempResources) {
+      tempResource.delete();
+    }
+    agentProps = null;
+  }
+
+  /**
+   * File channel in/out test. Verifies that all events inserted into the
+   * file channel are received by the sink in order.
+   * <p>
+   * The EXEC source creates 100 events where the event bodies have
+   * sequential numbers. The source puts those events into the file channel,
+   * and the FILE_ROLL The sink is expected to take all 100 events in FIFO
+   * order.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testInOut() throws Exception {
+    LOGGER.debug("testInOut() started.");
+
+
+    DockerInstall.getInstance().startAgent("a1", agentProps, tempResources);
+    TimeUnit.SECONDS.sleep(10); // Wait for source and sink to finish
+    // TODO make this more deterministic
+
+    /* Create expected output */
+    StringBuffer sb = new StringBuffer();
+    for (int i = 1; i <= 100; i++) {
+      sb.append(i).append("\n");
+    }
+    String expectedOutput = sb.toString();
+    LOGGER.info("Created expected output: " + expectedOutput);
+
+    /* Create actual output file */
+
+    File[] sinkOutputDirChildren = sinkOutputDir.listFiles();
+    // Only 1 file should be in FILE_ROLL sink's dir (rolling is disabled)
+    Assert.assertEquals("Expected FILE_ROLL sink's dir to have only 1 child," +
+                        " but found " + sinkOutputDirChildren.length + " children.",
+                        1, sinkOutputDirChildren.length);
+    File actualOutput = sinkOutputDirChildren[0];
+
+    if (!Files.toString(actualOutput, Charsets.UTF_8).equals(expectedOutput)) {
+      LOGGER.error("Actual output doesn't match expected output.\n");
+      throw new AssertionError("FILE_ROLL sink's actual output doesn't " +
+                               "match expected output.");
+    }
+
+    LOGGER.debug("testInOut() ended.");
+  }
+}