You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by mo...@apache.org on 2023/01/12 08:32:47 UTC
[incubator-streampark] branch dev updated: [Bug] get job log on k8s bug fixed (#2249)
This is an automated email from the ASF dual-hosted git repository.
monster pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new 4e7c27586 [Bug] get job log on k8s bug fixed (#2249)
4e7c27586 is described below
commit 4e7c275861919d8646323ac256d0bbe9fad1300c
Author: benjobs <be...@apache.org>
AuthorDate: Thu Jan 12 16:32:39 2023 +0800
[Bug] get job log on k8s bug fixed (#2249)
---
.../console/core/service/LogClientService.java | 4 +-
.../core/service/impl/ApplicationServiceImpl.java | 34 +++++---
.../kubernetes/watcher/FlinkJobStatusWatcher.scala | 93 +++++++++++-----------
3 files changed, 68 insertions(+), 63 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LogClientService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LogClientService.java
index c9525ab64..e60a5115c 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LogClientService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LogClientService.java
@@ -23,7 +23,6 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.io.File;
-import java.io.FileNotFoundException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
@@ -44,9 +43,8 @@ public class LogClientService {
lines.forEach(line -> builder.append(line).append("\r\n"));
return builder.toString();
}
- } else {
- throw new FileNotFoundException("file path: " + path + " not exists ");
}
+ return null;
} catch (Exception e) {
throw new ApiDetailException("roll view log error: " + e);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index 778dcfe67..ad0ff6626 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -587,6 +587,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
Application application = getById(id);
AssertUtils.state(application != null);
if (ExecutionMode.isKubernetesMode(application.getExecutionModeEnum())) {
+
CompletableFuture<String> future =
CompletableFuture.supplyAsync(
() ->
@@ -595,24 +596,31 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
application.getJobName(),
application.getJobId()));
- return CompletableFutureUtils.supplyTimeout(
- future,
- 5,
- TimeUnit.SECONDS,
- success -> success,
- exception -> {
+ return future
+ .exceptionally(
+ e -> {
String errorLog =
String.format(
- "%s/%s_err.log", WebUtils.getAppTempDir(), application.getJobId());
+ "%s/%s_err.log",
+ WebUtils.getAppTempDir().getAbsolutePath(), application.getJobId());
File file = new File(errorLog);
- if (file.exists()) {
- return errorLog;
- } else {
- throw new ApiDetailException("get k8s job log failed: " + exception.getMessage());
+ if (file.exists() && file.isFile()) {
+ return file.getAbsolutePath();
+ }
+ return null;
+ })
+ .thenApply(
+ path -> {
+ if (!future.isDone()) {
+ future.cancel(true);
+ }
+ if (path != null) {
+ return logClient.rollViewLog(path, offset, limit);
}
+ return null;
})
- .thenApply(path -> logClient.rollViewLog(path, offset, limit))
- .get();
+ .toCompletableFuture()
+ .get(5, TimeUnit.SECONDS);
} else {
throw new ApiAlertException(
"job executionMode must be kubernetes-session|kubernetes-application");
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
index edd41d727..bc204959a 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
@@ -31,7 +31,6 @@ import scala.util.{Failure, Success, Try}
import com.google.common.base.Charsets
import com.google.common.io.Files
-import org.apache.commons.collections.CollectionUtils
import org.apache.flink.core.fs.Path
import org.apache.flink.runtime.history.FsJobArchivist
import org.apache.hc.client5.http.fluent.Request
@@ -56,8 +55,8 @@ import org.apache.streampark.flink.kubernetes.model._
*/
@ThreadSafe
class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig = JobStatusWatcherConfig.defaultConf)(
- implicit val watchController: FlinkK8sWatchController,
- implicit val eventBus: ChangeEventBus) extends Logger with FlinkWatcher {
+ implicit val watchController: FlinkK8sWatchController,
+ implicit val eventBus: ChangeEventBus) extends Logger with FlinkWatcher {
private val trackTaskExecPool = Executors.newWorkStealingPool()
implicit private val trackTaskExecutor: ExecutionContextExecutorService = ExecutionContext.fromExecutorService(trackTaskExecPool)
@@ -96,7 +95,7 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig = JobStatusWatcherConfi
*/
override def doWatch(): Unit = {
this.synchronized {
- logInfo("[FlinkJobStatusWatcher]: Status monitoring process begins - " + Thread.currentThread().getName)
+ logDebug("[FlinkJobStatusWatcher]: Status monitoring process begins - " + Thread.currentThread().getName)
// get all legal tracking ids
val trackIds = Try(watchController.getAllWatchingIds()).filter(_.nonEmpty).getOrElse(return
)
@@ -136,11 +135,11 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig = JobStatusWatcherConfi
// blocking until all future are completed or timeout is reached
Try(Await.ready(Future.sequence(tracksFuture), conf.requestTimeoutSec seconds))
.failed.map { _ =>
- logInfo(s"[FlinkJobStatusWatcher] tracking flink job status on kubernetes mode timeout," +
- s" limitSeconds=${conf.requestTimeoutSec}," +
- s" trackIds=${trackIds.mkString(",")}")
- }
- logInfo("[FlinkJobStatusWatcher]: End of status monitoring process - " + Thread.currentThread().getName)
+ logWarn(s"[FlinkJobStatusWatcher] tracking flink job status on kubernetes mode timeout," +
+ s" limitSeconds=${conf.requestTimeoutSec}," +
+ s" trackIds=${trackIds.mkString(",")}")
+ }
+ logDebug("[FlinkJobStatusWatcher]: End of status monitoring process - " + Thread.currentThread().getName)
}
}
@@ -308,8 +307,8 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig = JobStatusWatcherConfi
private[this] def inferSilentOrLostFromPreCache(preCache: JobStatusCV) = preCache match {
case preCache if preCache == null => FlinkJobState.SILENT
case preCache
- if preCache.jobState == FlinkJobState.SILENT &&
- System.currentTimeMillis() - preCache.pollAckTime >= conf.silentStateJobKeepTrackingSec * 1000 => FlinkJobState.LOST
+ if preCache.jobState == FlinkJobState.SILENT &&
+ System.currentTimeMillis() - preCache.pollAckTime >= conf.silentStateJobKeepTrackingSec * 1000 => FlinkJobState.LOST
case _ => FlinkJobState.SILENT
}
@@ -330,10 +329,10 @@ object FlinkJobStatusWatcher {
current match {
case FlinkJobState.LOST => if (effectEndStates.contains(current)) previous else FlinkJobState.TERMINATED
case FlinkJobState.POS_TERMINATED | FlinkJobState.TERMINATED => previous match {
- case FlinkJobState.CANCELLING => FlinkJobState.CANCELED
- case FlinkJobState.FAILING => FlinkJobState.FAILED
- case _ => if (current == FlinkJobState.POS_TERMINATED) FlinkJobState.FINISHED else FlinkJobState.TERMINATED
- }
+ case FlinkJobState.CANCELLING => FlinkJobState.CANCELED
+ case FlinkJobState.FAILING => FlinkJobState.FAILED
+ case _ => if (current == FlinkJobState.POS_TERMINATED) FlinkJobState.FINISHED else FlinkJobState.TERMINATED
+ }
case _ => current
}
}
@@ -433,41 +432,41 @@ private[kubernetes] object FlinkHistoryArchives {
def getJobStateFromArchiveFile(jobId: String): String = Try {
require(jobId != null, "[StreamPark] getJobStateFromArchiveFile: JobId cannot be null.")
val archivePath = new Path(Workspace.ARCHIVES_FILE_PATH, jobId)
- val archivedJson = FsJobArchivist.getArchivedJsons(archivePath)
- var state: String = FAILED_STATE
- if (CollectionUtils.isNotEmpty(archivedJson)) {
- archivedJson.foreach { a =>
- if (a.getPath == s"/jobs/$jobId/exceptions") {
- Try(parse(a.getJson)) match {
- case Success(ok) =>
- val log = (ok \ "root-exception").extractOpt[String].orNull
- if (log != null) {
- val path = KubernetesDeploymentHelper.getJobErrorLog(jobId)
- val file = new File(path)
- Files.asCharSink(file, Charsets.UTF_8).write(log)
- }
- case _ =>
- }
- } else if (a.getPath == "/jobs/overview") {
- Try(parse(a.getJson)) match {
- case Success(ok) =>
- ok \ "jobs" match {
- case JNothing | JNull =>
- case JArray(arr) =>
- arr.foreach(x => {
- val jid = (x \ "jid").extractOpt[String].orNull
- if (jid == jobId) {
- state = (x \ "state").extractOpt[String].orNull
- }
- })
- case _ =>
- }
- case Failure(_) =>
+ FsJobArchivist.getArchivedJsons(archivePath) match {
+ case r if r.isEmpty => FAILED_STATE
+ case r =>
+ r.foreach { a =>
+ if (a.getPath == s"/jobs/$jobId/exceptions") {
+ Try(parse(a.getJson)) match {
+ case Success(ok) =>
+ val log = (ok \ "root-exception").extractOpt[String].orNull
+ if (log != null) {
+ val path = KubernetesDeploymentHelper.getJobErrorLog(jobId)
+ val file = new File(path)
+ Files.asCharSink(file, Charsets.UTF_8).write(log)
+ }
+ case _ =>
+ }
+ } else if (a.getPath == "/jobs/overview") {
+ Try(parse(a.getJson)) match {
+ case Success(ok) =>
+ ok \ "jobs" match {
+ case JNothing | JNull =>
+ case JArray(arr) =>
+ arr.foreach(x => {
+ val jid = (x \ "jid").extractOpt[String].orNull
+ if (jid == jobId) {
+ return (x \ "state").extractOpt[String].orNull
+ }
+ })
+ case _ =>
+ }
+ case Failure(_) =>
+ }
}
}
- }
+ FAILED_STATE
}
- state
}.getOrElse(FAILED_STATE)
}