You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/08/02 08:01:32 UTC
[GitHub] [flink] zentol commented on a change in pull request #16166: [FLINK-23000] Add log bundler to collect and compress logs
zentol commented on a change in pull request #16166:
URL: https://github.com/apache/flink/pull/16166#discussion_r680722795
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogArchiver.java
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
+import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/** Simple util for creating a log archive. */
+public class LogArchiver {
+ private static final Logger log = LoggerFactory.getLogger(LogArchiver.class);
+ private static final Pattern namePattern = Pattern.compile("\\.([0-9]+)$");
+
+ private final TarArchiveOutputStream archiveOutputStream;
+ private final File archive;
+ private boolean isClosed;
+ private final Set<String> entryNames = new HashSet<>();
+
+ public LogArchiver(File file) throws IOException {
+ this.archive = file;
+
+ OutputStream fo =
+ Files.newOutputStream(
+ file.toPath(),
+ StandardOpenOption.CREATE,
+ StandardOpenOption.WRITE,
+ StandardOpenOption.TRUNCATE_EXISTING);
+ OutputStream gzo = new GzipCompressorOutputStream(fo);
+ this.archiveOutputStream = new TarArchiveOutputStream(gzo);
+ isClosed = false;
+ entryNames.clear();
+ }
+
+ public File getArchive() {
+ checkState(isClosed, "Can not download archive before it has been closed.");
+ return archive;
+ }
+
+ public synchronized void addArchiveEntry(
+ String entryName, InputStream input, long size, long modTime) throws IOException {
+ checkState(!isClosed, "Can not add archive entry to closed archiver");
+ String internalEntryName = entryName;
+ while (entryNames.contains(internalEntryName)) {
+ internalEntryName = getNextEntryName(internalEntryName);
+ }
+ log.debug("Adding archive entry name:{} size:{}", internalEntryName, size);
+ entryNames.add(internalEntryName);
+ TarArchiveEntry entry = new TarArchiveEntry(internalEntryName);
+ entry.setSize(size);
+ entry.setMode(TarArchiveEntry.DEFAULT_FILE_MODE);
+ entry.setModTime(modTime);
+ entry.setUserName("");
Review comment:
add a comment that is done because it defaults to the `user.name` system property.
What does the userName actually do?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogBundlerHandler.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rest.handler.AbstractHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerActionQueryParameter;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerMessageParameters;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerStatus;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.ThreadDumpInfo;
+import org.apache.flink.runtime.taskexecutor.FileType;
+import org.apache.flink.runtime.util.JvmUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ThreadInfo;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler.getResourceManagerGateway;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Handler which creates a downloadable archive containing all local (JobManager) and remote
+ * (JobManager) log files.
+ */
+public class LogBundlerHandler
+ extends AbstractHandler<RestfulGateway, EmptyRequestBody, LogBundlerMessageParameters> {
+
+ // set high timeout for uploading files to the blob store
+ private final Time rpcTimeout = Time.minutes(1);
+ private final Object statusLock = new Object();
+ private final ScheduledExecutorService executor;
+ private final GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever;
+ @Nullable private final File localLogDir;
+ private final TransientBlobService transientBlobService;
+
+ @Nullable private LogArchiver logArchiver;
+
+ // null tmp dir indicates the bundler is disabled
+ @Nullable private final String tmpDir;
+
+ // messages for the REST API / frontend
+ private volatile String message = "";
+
+ // status of the bundler
+ @GuardedBy("statusLock")
+ private volatile Status status = Status.IDLE;
Review comment:
Why is this handler implemented with a custom state machine, instead of as async handler like the savepoint handler?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogBundlerHandler.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rest.handler.AbstractHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerActionQueryParameter;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerMessageParameters;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerStatus;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.ThreadDumpInfo;
+import org.apache.flink.runtime.taskexecutor.FileType;
+import org.apache.flink.runtime.util.JvmUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ThreadInfo;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler.getResourceManagerGateway;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Handler which creates a downloadable archive containing all local (JobManager) and remote
+ * (JobManager) log files.
+ */
+public class LogBundlerHandler
+ extends AbstractHandler<RestfulGateway, EmptyRequestBody, LogBundlerMessageParameters> {
+
+ // set high timeout for uploading files to the blob store
+ private final Time rpcTimeout = Time.minutes(1);
+ private final Object statusLock = new Object();
+ private final ScheduledExecutorService executor;
+ private final GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever;
+ @Nullable private final File localLogDir;
+ private final TransientBlobService transientBlobService;
+
+ @Nullable private LogArchiver logArchiver;
+
+ // null tmp dir indicates the bundler is disabled
+ @Nullable private final String tmpDir;
+
+ // messages for the REST API / frontend
+ private volatile String message = "";
+
+ // status of the bundler
+ @GuardedBy("statusLock")
+ private volatile Status status = Status.IDLE;
+
+ /** States of the log bundler. */
+ public enum Status {
+ IDLE,
+ PROCESSING,
+ BUNDLE_READY,
+ BUNDLE_FAILED
+ }
+
+ public LogBundlerHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ UntypedResponseMessageHeaders<EmptyRequestBody, LogBundlerMessageParameters>
+ messageHeaders,
+ ScheduledExecutorService executor,
+ Configuration clusterConfiguration,
+ @Nullable File localLogDir,
+ GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
+ TransientBlobService transientBlobService) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+ this.executor = executor;
+ this.resourceManagerGatewayRetriever = resourceManagerGatewayRetriever;
+ this.transientBlobService = transientBlobService;
+ this.localLogDir = localLogDir;
+ String[] tmpDirs = ConfigurationUtils.parseTempDirectories(clusterConfiguration);
+ if (tmpDirs.length == 0) {
+ this.message = "No tmp directory available for log bundler";
+ log.error(message);
+ this.tmpDir = null;
Review comment:
We don't need to handle this case. If no tmpDir was configured the cluster won't even start up.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogBundlerHandler.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rest.handler.AbstractHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerActionQueryParameter;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerMessageParameters;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerStatus;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.ThreadDumpInfo;
+import org.apache.flink.runtime.taskexecutor.FileType;
+import org.apache.flink.runtime.util.JvmUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ThreadInfo;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler.getResourceManagerGateway;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Handler which creates a downloadable archive containing all local (JobManager) and remote
+ * (JobManager) log files.
+ */
+public class LogBundlerHandler
+ extends AbstractHandler<RestfulGateway, EmptyRequestBody, LogBundlerMessageParameters> {
+
+ // set high timeout for uploading files to the blob store
+ private final Time rpcTimeout = Time.minutes(1);
Review comment:
this should be based in some form on the configured rpc timeout.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/logbundler/LogArchiverTest.java
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.util.TestLogger;
+
+import org.apache.commons.compress.archivers.ArchiveEntry;
+import org.apache.commons.compress.archivers.ArchiveInputStream;
+import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
+import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
+
+/** Tests for {@link LogArchiver}. */
+public class LogArchiverTest extends TestLogger {
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testLogArchiver() throws IOException {
+ File archive = temporaryFolder.newFile();
+ LogArchiver la = new LogArchiver(archive);
+
+ la.addArchiveEntry("file", createFileContainingString("first"));
+ la.addArchiveEntry("file", createFileContainingString("second"));
+ la.addArchiveEntry("file", createFileContainingString("third"));
+
+ la.closeArchive();
+
+ assertThat(la.getArchive(), is(archive));
+
+ List<String> filesInArchive = new ArrayList<>();
+ try (InputStream bi = new FileInputStream(archive);
+ InputStream gzi = new GzipCompressorInputStream(bi);
+ ArchiveInputStream o = new TarArchiveInputStream(gzi)) {
+ ArchiveEntry entry = null;
+ while ((entry = o.getNextEntry()) != null) {
+ filesInArchive.add(entry.getName());
+ assertThat(entry.isDirectory(), is(false));
+ assertThat(entry.getSize(), greaterThan(0L));
+ }
+ }
+ assertThat(filesInArchive, hasSize(3));
+ assertThat(filesInArchive, contains("file", "file.1", "file.2"));
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testGetWhenNotClosed() throws IOException {
+ LogArchiver la = new LogArchiver(temporaryFolder.newFile());
+ la.getArchive();
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testAddWhenClosed() throws IOException {
+ LogArchiver la = new LogArchiver(temporaryFolder.newFile());
+ la.closeArchive();
+ la.addArchiveEntry("file", createFileContainingString("test"));
+ }
+
+ @Test
+ public void testDestroyDeletesFile() throws IOException {
+ File f = temporaryFolder.newFile();
+ LogArchiver la = new LogArchiver(f);
+ assertThat(f.exists(), is(true));
+ la.destroy();
+ assertThat(f.exists(), is(false));
+ }
+
+ private File createFileContainingString(String s) throws IOException {
+ File f = temporaryFolder.newFile();
+ Files.write(f.toPath(), s.getBytes(StandardCharsets.UTF_8), StandardOpenOption.CREATE);
+ return f;
+ }
+
+ @Test
+ public void testGetNextEntryName() {
+ assertThat(
+ LogArchiver.getNextEntryName("jobmanager-test.log"), is("jobmanager-test.log.1"));
+ assertThat(
+ LogArchiver.getNextEntryName("jobmanager-test.log.3"), is("jobmanager-test.log.4"));
+
+ assertThat(
+ LogArchiver.getNextEntryName("jobmanager-test.log.3.4.2.3"),
+ is("jobmanager-test.log.3.4.2.4"));
+ }
+
+ @Test
+ public void testCopy_full() throws IOException {
+ byte[] input = new byte[10000];
+ InputStream inputStream = new ByteArrayInputStream(input);
+ ByteArrayOutputStream bao = new ByteArrayOutputStream();
+ LogArchiver.copyWithLimit(inputStream, bao, input.length);
+
+ assertThat(bao.size(), equalTo(input.length));
+ }
+
+ @Test
+ public void testCopy_limited() throws IOException {
+ int limit = 100;
+ byte[] input = new byte[10000];
+ InputStream inputStream = new ByteArrayInputStream(input);
+ ByteArrayOutputStream bao = new ByteArrayOutputStream();
+ LogArchiver.copyWithLimit(inputStream, bao, limit);
+
+ assertThat(bao.size(), equalTo(limit));
+ }
+
+ @Test
+ public void testCopy_limited2() throws IOException {
Review comment:
what is this test for?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogBundlerHandler.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rest.handler.AbstractHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerActionQueryParameter;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerMessageParameters;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerStatus;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.ThreadDumpInfo;
+import org.apache.flink.runtime.taskexecutor.FileType;
+import org.apache.flink.runtime.util.JvmUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ThreadInfo;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler.getResourceManagerGateway;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Handler which creates a downloadable archive containing all local (JobManager) and remote
+ * (JobManager) log files.
+ */
+public class LogBundlerHandler
+ extends AbstractHandler<RestfulGateway, EmptyRequestBody, LogBundlerMessageParameters> {
+
+ // set high timeout for uploading files to the blob store
+ private final Time rpcTimeout = Time.minutes(1);
+ private final Object statusLock = new Object();
+ private final ScheduledExecutorService executor;
+ private final GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever;
+ @Nullable private final File localLogDir;
+ private final TransientBlobService transientBlobService;
+
+ @Nullable private LogArchiver logArchiver;
+
+ // null tmp dir indicates the bundler is disabled
+ @Nullable private final String tmpDir;
+
+ // messages for the REST API / frontend
+ private volatile String message = "";
+
+ // status of the bundler
+ @GuardedBy("statusLock")
+ private volatile Status status = Status.IDLE;
+
+ /** States of the log bundler. */
+ public enum Status {
+ IDLE,
+ PROCESSING,
+ BUNDLE_READY,
+ BUNDLE_FAILED
+ }
+
+ public LogBundlerHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ UntypedResponseMessageHeaders<EmptyRequestBody, LogBundlerMessageParameters>
+ messageHeaders,
+ ScheduledExecutorService executor,
+ Configuration clusterConfiguration,
+ @Nullable File localLogDir,
+ GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
+ TransientBlobService transientBlobService) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+ this.executor = executor;
+ this.resourceManagerGatewayRetriever = resourceManagerGatewayRetriever;
+ this.transientBlobService = transientBlobService;
+ this.localLogDir = localLogDir;
+ String[] tmpDirs = ConfigurationUtils.parseTempDirectories(clusterConfiguration);
+ if (tmpDirs.length == 0) {
+ this.message = "No tmp directory available for log bundler";
+ log.error(message);
+ this.tmpDir = null;
+ } else {
+ this.tmpDir = tmpDirs[0];
+ }
+ }
+
+ @Override
+ protected CompletableFuture<Void> respondToRequest(
+ ChannelHandlerContext ctx,
+ HttpRequest httpRequest,
+ HandlerRequest<EmptyRequestBody, LogBundlerMessageParameters> handlerRequest,
+ RestfulGateway gateway)
+ throws RestHandlerException {
+ List<String> queryParams =
+ handlerRequest.getQueryParameter(LogBundlerActionQueryParameter.class);
+ if (!queryParams.isEmpty()) {
+ if (tmpDir == null) {
+ throw new RestHandlerException(
+ this.message, HttpResponseStatus.INTERNAL_SERVER_ERROR);
+ }
+ synchronized (statusLock) {
+ final String action = queryParams.get(0);
+ if ("download".equals(action)) {
Review comment:
Move download/trigger into an enum that is also used in the headers.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogArchiver.java
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
+import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/** Simple util for creating a log archive. */
+public class LogArchiver {
+ private static final Logger log = LoggerFactory.getLogger(LogArchiver.class);
+ private static final Pattern namePattern = Pattern.compile("\\.([0-9]+)$");
+
+ private final TarArchiveOutputStream archiveOutputStream;
+ private final File archive;
+ private boolean isClosed;
+ private final Set<String> entryNames = new HashSet<>();
+
+ public LogArchiver(File file) throws IOException {
+ this.archive = file;
+
+ OutputStream fo =
+ Files.newOutputStream(
+ file.toPath(),
+ StandardOpenOption.CREATE,
+ StandardOpenOption.WRITE,
+ StandardOpenOption.TRUNCATE_EXISTING);
+ OutputStream gzo = new GzipCompressorOutputStream(fo);
+ this.archiveOutputStream = new TarArchiveOutputStream(gzo);
+ isClosed = false;
+ entryNames.clear();
+ }
+
+ public File getArchive() {
+ checkState(isClosed, "Can not download archive before it has been closed.");
+ return archive;
+ }
+
+ public synchronized void addArchiveEntry(
+ String entryName, InputStream input, long size, long modTime) throws IOException {
+ checkState(!isClosed, "Can not add archive entry to closed archiver");
+ String internalEntryName = entryName;
+ while (entryNames.contains(internalEntryName)) {
+ internalEntryName = getNextEntryName(internalEntryName);
+ }
+ log.debug("Adding archive entry name:{} size:{}", internalEntryName, size);
+ entryNames.add(internalEntryName);
+ TarArchiveEntry entry = new TarArchiveEntry(internalEntryName);
+ entry.setSize(size);
+ entry.setMode(TarArchiveEntry.DEFAULT_FILE_MODE);
+ entry.setModTime(modTime);
+ entry.setUserName("");
+ try {
+ archiveOutputStream.putArchiveEntry(entry);
+ // copy with limit: the input log file could have additional data, which would confuse
+ // the archiver
Review comment:
Can you expand on this?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogBundlerHandler.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rest.handler.AbstractHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerActionQueryParameter;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerMessageParameters;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerStatus;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.ThreadDumpInfo;
+import org.apache.flink.runtime.taskexecutor.FileType;
+import org.apache.flink.runtime.util.JvmUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ThreadInfo;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler.getResourceManagerGateway;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Handler which creates a downloadable archive containing all local (JobManager) and remote
+ * (JobManager) log files.
+ */
+public class LogBundlerHandler
+ extends AbstractHandler<RestfulGateway, EmptyRequestBody, LogBundlerMessageParameters> {
+
+ // set high timeout for uploading files to the blob store
+ private final Time rpcTimeout = Time.minutes(1);
+ private final Object statusLock = new Object();
+ private final ScheduledExecutorService executor;
+ private final GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever;
+ @Nullable private final File localLogDir;
+ private final TransientBlobService transientBlobService;
+
+ @Nullable private LogArchiver logArchiver;
+
+ // null tmp dir indicates the bundler is disabled
+ @Nullable private final String tmpDir;
+
+ // messages for the REST API / frontend
+ private volatile String message = "";
+
+ // status of the bundler
+ @GuardedBy("statusLock")
+ private volatile Status status = Status.IDLE;
+
+ /** States of the log bundler. */
+ public enum Status {
+ IDLE,
+ PROCESSING,
+ BUNDLE_READY,
+ BUNDLE_FAILED
+ }
+
+ public LogBundlerHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ UntypedResponseMessageHeaders<EmptyRequestBody, LogBundlerMessageParameters>
+ messageHeaders,
+ ScheduledExecutorService executor,
+ Configuration clusterConfiguration,
+ @Nullable File localLogDir,
+ GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
+ TransientBlobService transientBlobService) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+ this.executor = executor;
+ this.resourceManagerGatewayRetriever = resourceManagerGatewayRetriever;
+ this.transientBlobService = transientBlobService;
+ this.localLogDir = localLogDir;
+ String[] tmpDirs = ConfigurationUtils.parseTempDirectories(clusterConfiguration);
+ if (tmpDirs.length == 0) {
+ this.message = "No tmp directory available for log bundler";
+ log.error(message);
+ this.tmpDir = null;
+ } else {
+ this.tmpDir = tmpDirs[0];
+ }
+ }
+
+ @Override
+ protected CompletableFuture<Void> respondToRequest(
+ ChannelHandlerContext ctx,
+ HttpRequest httpRequest,
+ HandlerRequest<EmptyRequestBody, LogBundlerMessageParameters> handlerRequest,
+ RestfulGateway gateway)
+ throws RestHandlerException {
+ List<String> queryParams =
+ handlerRequest.getQueryParameter(LogBundlerActionQueryParameter.class);
+ if (!queryParams.isEmpty()) {
+ if (tmpDir == null) {
+ throw new RestHandlerException(
+ this.message, HttpResponseStatus.INTERNAL_SERVER_ERROR);
+ }
+ synchronized (statusLock) {
+ final String action = queryParams.get(0);
+ if ("download".equals(action)) {
+ if (status != Status.BUNDLE_READY) {
+ throw new RestHandlerException(
+ "There is no bundle ready to be downloaded",
+ HttpResponseStatus.BAD_REQUEST);
+ }
+ checkNotNull(
+ logArchiver,
+ "Assuming log archiver to be set in status " + Status.BUNDLE_READY);
+
+ try {
+ HandlerUtils.transferFile(ctx, logArchiver.getArchive(), httpRequest);
+ return CompletableFuture.completedFuture(null);
+ } catch (FlinkException e) {
+ log.warn("Error while transferring file", e);
+ }
+ } else if ("trigger".equals(action)) {
+ if (status == Status.PROCESSING) {
+ throw new RestHandlerException(
+ "Unable to trigger log bundling while in status "
+ + Status.PROCESSING,
+ HttpResponseStatus.BAD_REQUEST);
+ }
+ status = Status.PROCESSING;
+ executor.execute(this::collectAndCompressLogs);
+ } else {
+ log.warn("Unknown action passed: '{}'", action);
+ }
+ }
+ }
+
+ // access status outside of lock to respond to status queries while processing.
+ return HandlerUtils.sendResponse(
+ ctx,
+ httpRequest,
+ new LogBundlerStatus(status, message),
+ HttpResponseStatus.OK,
+ responseHeaders);
+ }
+
+ private void collectAndCompressLogs() {
+ synchronized (statusLock) {
Review comment:
This locking kind of voids the benefit of using an executor, because as long as this operation is running any request to the handler results in a blocked thread.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogBundlerHandler.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rest.handler.AbstractHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerActionQueryParameter;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerMessageParameters;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerStatus;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.ThreadDumpInfo;
+import org.apache.flink.runtime.taskexecutor.FileType;
+import org.apache.flink.runtime.util.JvmUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ThreadInfo;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler.getResourceManagerGateway;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Handler which creates a downloadable archive containing all local (JobManager) and remote
+ * (JobManager) log files.
+ */
+public class LogBundlerHandler
+ extends AbstractHandler<RestfulGateway, EmptyRequestBody, LogBundlerMessageParameters> {
+
+ // set high timeout for uploading files to the blob store
+ private final Time rpcTimeout = Time.minutes(1);
+ private final Object statusLock = new Object();
+ private final ScheduledExecutorService executor;
+ private final GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever;
+ @Nullable private final File localLogDir;
+ private final TransientBlobService transientBlobService;
+
+ @Nullable private LogArchiver logArchiver;
+
+ // null tmp dir indicates the bundler is disabled
+ @Nullable private final String tmpDir;
+
+ // messages for the REST API / frontend
+ private volatile String message = "";
+
+ // status of the bundler
+ @GuardedBy("statusLock")
+ private volatile Status status = Status.IDLE;
+
+ /** States of the log bundler. */
+ public enum Status {
+ IDLE,
+ PROCESSING,
+ BUNDLE_READY,
+ BUNDLE_FAILED
+ }
+
+ public LogBundlerHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ UntypedResponseMessageHeaders<EmptyRequestBody, LogBundlerMessageParameters>
+ messageHeaders,
+ ScheduledExecutorService executor,
+ Configuration clusterConfiguration,
+ @Nullable File localLogDir,
+ GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
+ TransientBlobService transientBlobService) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+ this.executor = executor;
+ this.resourceManagerGatewayRetriever = resourceManagerGatewayRetriever;
+ this.transientBlobService = transientBlobService;
+ this.localLogDir = localLogDir;
+ String[] tmpDirs = ConfigurationUtils.parseTempDirectories(clusterConfiguration);
+ if (tmpDirs.length == 0) {
+ this.message = "No tmp directory available for log bundler";
+ log.error(message);
+ this.tmpDir = null;
+ } else {
+ this.tmpDir = tmpDirs[0];
+ }
+ }
+
+ @Override
+ protected CompletableFuture<Void> respondToRequest(
+ ChannelHandlerContext ctx,
+ HttpRequest httpRequest,
+ HandlerRequest<EmptyRequestBody, LogBundlerMessageParameters> handlerRequest,
+ RestfulGateway gateway)
+ throws RestHandlerException {
+ List<String> queryParams =
+ handlerRequest.getQueryParameter(LogBundlerActionQueryParameter.class);
+ if (!queryParams.isEmpty()) {
+ if (tmpDir == null) {
+ throw new RestHandlerException(
+ this.message, HttpResponseStatus.INTERNAL_SERVER_ERROR);
+ }
+ synchronized (statusLock) {
+ final String action = queryParams.get(0);
+ if ("download".equals(action)) {
+ if (status != Status.BUNDLE_READY) {
+ throw new RestHandlerException(
+ "There is no bundle ready to be downloaded",
+ HttpResponseStatus.BAD_REQUEST);
+ }
+ checkNotNull(
+ logArchiver,
+ "Assuming log archiver to be set in status " + Status.BUNDLE_READY);
+
+ try {
+ HandlerUtils.transferFile(ctx, logArchiver.getArchive(), httpRequest);
+ return CompletableFuture.completedFuture(null);
+ } catch (FlinkException e) {
+ log.warn("Error while transferring file", e);
+ }
+ } else if ("trigger".equals(action)) {
+ if (status == Status.PROCESSING) {
+ throw new RestHandlerException(
+ "Unable to trigger log bundling while in status "
+ + Status.PROCESSING,
+ HttpResponseStatus.BAD_REQUEST);
+ }
+ status = Status.PROCESSING;
+ executor.execute(this::collectAndCompressLogs);
+ } else {
+ log.warn("Unknown action passed: '{}'", action);
+ }
+ }
+ }
+
+ // access status outside of lock to respond to status queries while processing.
+ return HandlerUtils.sendResponse(
+ ctx,
+ httpRequest,
+ new LogBundlerStatus(status, message),
+ HttpResponseStatus.OK,
+ responseHeaders);
+ }
+
+ private void collectAndCompressLogs() {
+ synchronized (statusLock) {
+ try {
+ checkState(status == Status.PROCESSING, "Not in status PROCESSING");
+ if (logArchiver != null) {
+ logArchiver.destroy();
+ }
+ final long creationTime = System.currentTimeMillis();
+ File archive =
+ new File(tmpDir + File.separator + creationTime + "-flink-log-bundle.tgz");
+ logArchiver = new LogArchiver(archive);
+ archive.deleteOnExit();
+
+ collectLocalLogs();
+ collectLocalThreacollectLocalLogs() dDump();
+ collectTaskManagerLogsAndThreadDumps();
+
+ status = Status.BUNDLE_READY;
+ message = "Created bundle at " + new Date(creationTime);
+ } catch (Throwable throwable) {
+ status = Status.BUNDLE_FAILED;
+ message = "Error while creating bundle " + throwable.getMessage();
+ log.warn("Error while creating bundle", throwable);
+ } finally {
+ try {
+ if (logArchiver != null) {
+ logArchiver.closeArchive();
+ }
+ } catch (IOException e) {
+ log.warn("Error while closing log archive", e);
+ }
+ }
+ }
+ }
+
+ private void collectLocalThreadDump() {
+ Collection<ThreadInfo> threadDump = JvmUtils.createThreadDump();
+ final Collection<ThreadDumpInfo.ThreadInfo> threadInfos =
+ threadDump.stream()
+ .map(
+ threadInfo ->
+ ThreadDumpInfo.ThreadInfo.create(
+ threadInfo.getThreadName(), threadInfo.toString()))
+ .collect(Collectors.toList());
+ addThreadDumps("jobmanager", threadInfos);
+ }
+
+ private void collectTaskManagerLogsAndThreadDumps()
+ throws RestHandlerException, ExecutionException, InterruptedException {
+ final ResourceManagerGateway resourceManagerGateway =
+ getResourceManagerGateway(resourceManagerGatewayRetriever);
+ Collection<TaskManagerInfo> taskManagers =
+ resourceManagerGateway.requestTaskManagerInfo(rpcTimeout).get();
+ // for this asynchronous collection of data, logArchiver.addArchiveEntry must be serialized
+ Collection<CompletableFuture<Void>> collectionTaskFutures = new ArrayList<>();
+ for (TaskManagerInfo taskManagerInfo : taskManagers) {
+ CompletableFuture<Void> collectLogs =
+ requestLogs(resourceManagerGateway, taskManagerInfo)
+ .thenAccept(
+ maybeLogs -> maybeLogs.ifPresent(this::addTaskManagerLogFile));
+ collectionTaskFutures.add(collectLogs);
+
+ CompletableFuture<Void> collectThreaddumps =
+ resourceManagerGateway
+ .requestThreadDump(taskManagerInfo.getResourceId(), rpcTimeout)
+ .thenAccept(
+ threadDumpInfo ->
+ addThreadDumps(
+ taskManagerInfo
+ .getResourceId()
+ .getResourceIdString(),
+ threadDumpInfo.getThreadInfos()));
+ collectionTaskFutures.add(collectThreaddumps);
+ }
+ // wait for all operations to complete
+ FutureUtils.combineAll(collectionTaskFutures).get();
+ }
+
+ private void addThreadDumps(
+ String name, Collection<ThreadDumpInfo.ThreadInfo> threadDumpInfos) {
+ checkNotNull(logArchiver, "Assuming log archiver to be set");
+ StringBuffer sb = new StringBuffer();
+ for (ThreadDumpInfo.ThreadInfo threadInfo : threadDumpInfos) {
+ sb.append(threadInfo.toString());
+ }
+
+ byte[] entry = sb.toString().getBytes(StandardCharsets.UTF_8);
+ try (InputStream input = new ByteArrayInputStream(entry)) {
+ logArchiver.addArchiveEntry(
+ name + ".threaddump", input, entry.length, System.currentTimeMillis());
+ } catch (IOException e) {
+ log.warn("Error while collecting thread dumps for {}", name, e);
+ }
+ }
+
+ private CompletableFuture<Optional<TaskManagerLogAndId>> requestLogs(
+ ResourceManagerGateway resourceManagerGateway, TaskManagerInfo taskManager) {
+ return resourceManagerGateway
+ .requestTaskManagerFileUploadByType(
+ taskManager.getResourceId(), FileType.LOG, rpcTimeout)
+ .thenApplyAsync(
+ tmLogBlobKey -> {
+ try {
+ return Optional.of(
+ new TaskManagerLogAndId(
+ taskManager.getResourceId(),
+ transientBlobService.getFile(tmLogBlobKey)));
+ } catch (IOException e) {
+ log.warn("Error while retrieving log from TaskManager", e);
+ return Optional.empty();
+ }
+ },
+ executor);
+ }
+
+ private void addTaskManagerLogFile(TaskManagerLogAndId logFile) {
+ checkNotNull(logArchiver, "Assuming log archiver to be set");
+ try {
+ logArchiver.addArchiveEntry(
Review comment:
this should run in a separate executor to ensure the REST API remains responsive.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/logbundler/LogBundlerHandler.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.flink.runtime.rest.handler.logbundler;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rest.handler.AbstractHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.UntypedResponseMessageHeaders;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerActionQueryParameter;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerMessageParameters;
+import org.apache.flink.runtime.rest.messages.logbundler.LogBundlerStatus;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.ThreadDumpInfo;
+import org.apache.flink.runtime.taskexecutor.FileType;
+import org.apache.flink.runtime.util.JvmUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ThreadInfo;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler.getResourceManagerGateway;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Handler which creates a downloadable archive containing all local (JobManager) and remote
+ * (JobManager) log files.
+ */
+public class LogBundlerHandler
+ extends AbstractHandler<RestfulGateway, EmptyRequestBody, LogBundlerMessageParameters> {
+
+ // set high timeout for uploading files to the blob store
+ private final Time rpcTimeout = Time.minutes(1);
+ private final Object statusLock = new Object();
+ private final ScheduledExecutorService executor;
+ private final GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever;
+ @Nullable private final File localLogDir;
+ private final TransientBlobService transientBlobService;
+
+ @Nullable private LogArchiver logArchiver;
+
+ // null tmp dir indicates the bundler is disabled
+ @Nullable private final String tmpDir;
+
+ // messages for the REST API / frontend
+ private volatile String message = "";
+
+ // status of the bundler
+ @GuardedBy("statusLock")
+ private volatile Status status = Status.IDLE;
+
+ /** States of the log bundler. */
+ public enum Status {
+ IDLE,
+ PROCESSING,
+ BUNDLE_READY,
+ BUNDLE_FAILED
+ }
+
+ public LogBundlerHandler(
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ Map<String, String> responseHeaders,
+ UntypedResponseMessageHeaders<EmptyRequestBody, LogBundlerMessageParameters>
+ messageHeaders,
+ ScheduledExecutorService executor,
+ Configuration clusterConfiguration,
+ @Nullable File localLogDir,
+ GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
+ TransientBlobService transientBlobService) {
+ super(leaderRetriever, timeout, responseHeaders, messageHeaders);
+ this.executor = executor;
+ this.resourceManagerGatewayRetriever = resourceManagerGatewayRetriever;
+ this.transientBlobService = transientBlobService;
+ this.localLogDir = localLogDir;
+ String[] tmpDirs = ConfigurationUtils.parseTempDirectories(clusterConfiguration);
+ if (tmpDirs.length == 0) {
+ this.message = "No tmp directory available for log bundler";
+ log.error(message);
+ this.tmpDir = null;
+ } else {
+ this.tmpDir = tmpDirs[0];
+ }
+ }
+
+ @Override
+ protected CompletableFuture<Void> respondToRequest(
+ ChannelHandlerContext ctx,
+ HttpRequest httpRequest,
+ HandlerRequest<EmptyRequestBody, LogBundlerMessageParameters> handlerRequest,
+ RestfulGateway gateway)
+ throws RestHandlerException {
+ List<String> queryParams =
+ handlerRequest.getQueryParameter(LogBundlerActionQueryParameter.class);
+ if (!queryParams.isEmpty()) {
+ if (tmpDir == null) {
+ throw new RestHandlerException(
+ this.message, HttpResponseStatus.INTERNAL_SERVER_ERROR);
+ }
+ synchronized (statusLock) {
+ final String action = queryParams.get(0);
+ if ("download".equals(action)) {
+ if (status != Status.BUNDLE_READY) {
+ throw new RestHandlerException(
+ "There is no bundle ready to be downloaded",
+ HttpResponseStatus.BAD_REQUEST);
+ }
+ checkNotNull(
+ logArchiver,
+ "Assuming log archiver to be set in status " + Status.BUNDLE_READY);
+
+ try {
+ HandlerUtils.transferFile(ctx, logArchiver.getArchive(), httpRequest);
+ return CompletableFuture.completedFuture(null);
+ } catch (FlinkException e) {
+ log.warn("Error while transferring file", e);
+ }
+ } else if ("trigger".equals(action)) {
+ if (status == Status.PROCESSING) {
+ throw new RestHandlerException(
+ "Unable to trigger log bundling while in status "
+ + Status.PROCESSING,
+ HttpResponseStatus.BAD_REQUEST);
+ }
+ status = Status.PROCESSING;
+ executor.execute(this::collectAndCompressLogs);
+ } else {
+ log.warn("Unknown action passed: '{}'", action);
+ }
+ }
+ }
+
+ // access status outside of lock to respond to status queries while processing.
+ return HandlerUtils.sendResponse(
+ ctx,
+ httpRequest,
+ new LogBundlerStatus(status, message),
+ HttpResponseStatus.OK,
+ responseHeaders);
+ }
+
+ private void collectAndCompressLogs() {
+ synchronized (statusLock) {
+ try {
+ checkState(status == Status.PROCESSING, "Not in status PROCESSING");
+ if (logArchiver != null) {
+ logArchiver.destroy();
+ }
+ final long creationTime = System.currentTimeMillis();
+ File archive =
+ new File(tmpDir + File.separator + creationTime + "-flink-log-bundle.tgz");
+ logArchiver = new LogArchiver(archive);
+ archive.deleteOnExit();
+
+ collectLocalLogs();
+ collectLocalThreacollectLocalLogs() dDump();
+ collectTaskManagerLogsAndThreadDumps();
+
+ status = Status.BUNDLE_READY;
+ message = "Created bundle at " + new Date(creationTime);
+ } catch (Throwable throwable) {
+ status = Status.BUNDLE_FAILED;
+ message = "Error while creating bundle " + throwable.getMessage();
+ log.warn("Error while creating bundle", throwable);
+ } finally {
+ try {
+ if (logArchiver != null) {
+ logArchiver.closeArchive();
+ }
+ } catch (IOException e) {
+ log.warn("Error while closing log archive", e);
+ }
+ }
+ }
+ }
+
+ private void collectLocalThreadDump() {
+ Collection<ThreadInfo> threadDump = JvmUtils.createThreadDump();
+ final Collection<ThreadDumpInfo.ThreadInfo> threadInfos =
+ threadDump.stream()
+ .map(
+ threadInfo ->
+ ThreadDumpInfo.ThreadInfo.create(
+ threadInfo.getThreadName(), threadInfo.toString()))
+ .collect(Collectors.toList());
+ addThreadDumps("jobmanager", threadInfos);
+ }
+
+ private void collectTaskManagerLogsAndThreadDumps()
+ throws RestHandlerException, ExecutionException, InterruptedException {
+ final ResourceManagerGateway resourceManagerGateway =
+ getResourceManagerGateway(resourceManagerGatewayRetriever);
+ Collection<TaskManagerInfo> taskManagers =
+ resourceManagerGateway.requestTaskManagerInfo(rpcTimeout).get();
+ // for this asynchronous collection of data, logArchiver.addArchiveEntry must be serialized
+ Collection<CompletableFuture<Void>> collectionTaskFutures = new ArrayList<>();
+ for (TaskManagerInfo taskManagerInfo : taskManagers) {
+ CompletableFuture<Void> collectLogs =
+ requestLogs(resourceManagerGateway, taskManagerInfo)
+ .thenAccept(
+ maybeLogs -> maybeLogs.ifPresent(this::addTaskManagerLogFile));
+ collectionTaskFutures.add(collectLogs);
+
+ CompletableFuture<Void> collectThreaddumps =
+ resourceManagerGateway
+ .requestThreadDump(taskManagerInfo.getResourceId(), rpcTimeout)
+ .thenAccept(
+ threadDumpInfo ->
+ addThreadDumps(
+ taskManagerInfo
+ .getResourceId()
+ .getResourceIdString(),
+ threadDumpInfo.getThreadInfos()));
+ collectionTaskFutures.add(collectThreaddumps);
+ }
+ // wait for all operations to complete
+ FutureUtils.combineAll(collectionTaskFutures).get();
+ }
+
+ private void addThreadDumps(
+ String name, Collection<ThreadDumpInfo.ThreadInfo> threadDumpInfos) {
+ checkNotNull(logArchiver, "Assuming log archiver to be set");
+ StringBuffer sb = new StringBuffer();
+ for (ThreadDumpInfo.ThreadInfo threadInfo : threadDumpInfos) {
+ sb.append(threadInfo.toString());
+ }
+
+ byte[] entry = sb.toString().getBytes(StandardCharsets.UTF_8);
+ try (InputStream input = new ByteArrayInputStream(entry)) {
+ logArchiver.addArchiveEntry(
+ name + ".threaddump", input, entry.length, System.currentTimeMillis());
+ } catch (IOException e) {
+ log.warn("Error while collecting thread dumps for {}", name, e);
+ }
+ }
+
+ private CompletableFuture<Optional<TaskManagerLogAndId>> requestLogs(
+ ResourceManagerGateway resourceManagerGateway, TaskManagerInfo taskManager) {
+ return resourceManagerGateway
+ .requestTaskManagerFileUploadByType(
+ taskManager.getResourceId(), FileType.LOG, rpcTimeout)
+ .thenApplyAsync(
Review comment:
I assume this is an async operation such that subsequent steps run in the executor? If so, then I'd prefer making this explicitly when we setup the `addTaskManagerLogFile` call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org