Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat(core): refactor ExecutionLogService #5913

Open
wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

feat(core): refactor ExecutionLogService #5913

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 46 additions & 6 deletions core/src/main/java/io/kestra/core/services/ExecutionLogService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,10 +13,16 @@
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
* Service for fetching logs for from an execution.
*/
@Singleton
public class ExecutionLogService {

Expand All @@ -31,22 +37,23 @@ public Flux<Event<LogEntry>> streamExecutionLogs(final String tenantId,
final String executionId,
final Level minLevel) {

List<String> levels = LogEntry.findLevelsByMin(minLevel).stream().map(Enum::name).toList();

final AtomicReference<Runnable> disposable = new AtomicReference<>();

return Flux.<Event<LogEntry>>create(emitter -> {
// fetch repository first
fetchExecutionExecutionLogs(tenantId, executionId, minLevel, levels)
getExecutionLogs(tenantId, executionId, minLevel, List.of())
.forEach(logEntry -> emitter.next(Event.of(logEntry).id("progress")));

final List<String> levels = LogEntry.findLevelsByMin(minLevel).stream().map(Enum::name).toList();

// consume in realtime
disposable.set(this.logQueue.receive(either -> {
if (either.isRight()) {
return;
}

LogEntry current = either.getLeft();

if (current.getExecutionId() != null && current.getExecutionId().equals(executionId)) {
if (levels.contains(current.getLevel().name())) {
emitter.next(Event.of(current).id("progress"));
Expand All @@ -66,9 +73,42 @@ public Flux<Event<LogEntry>> streamExecutionLogs(final String tenantId,
});
}

public Stream<LogEntry> fetchExecutionExecutionLogs(String tenantId, String executionId, Level minLevel, List<String> levels) {
return logRepository.findByExecutionId(tenantId, executionId, minLevel, Pageable.UNPAGED)
public InputStream getExecutionLogsAsStream(final String tenantId,
final String executionId,
final Level minLevel,
final String taskRunId,
final List<String> taskIds,
final Integer attempt) {
List<LogEntry> logs = getExecutionLogs(tenantId, executionId, minLevel, taskRunId, taskIds, attempt);
return new ByteArrayInputStream(logs.stream().map(LogEntry::toPrettyString).collect(Collectors.joining("\n")).getBytes());
}

public List<LogEntry> getExecutionLogs(final String tenantId,
final String executionId,
final Level minLevel,
final String taskRunId,
final List<String> taskIds,
final Integer attempt) {
if (taskIds != null) {
return taskIds.size() == 1 ?
logRepository.findByExecutionIdAndTaskId(tenantId, executionId, taskIds.getFirst(), minLevel):
getExecutionLogs(tenantId, executionId, minLevel, taskIds).toList();
}

if (taskRunId != null) {
return attempt != null ?
logRepository.findByExecutionIdAndTaskRunIdAndAttempt(tenantId, executionId, taskRunId, minLevel, attempt) :
logRepository.findByExecutionIdAndTaskRunId(tenantId, executionId, taskRunId, minLevel);
}
return logRepository.findByExecutionId(tenantId, executionId, minLevel);
}

public Stream<LogEntry> getExecutionLogs(String tenantId,
String executionId,
Level minLevel,
List<String> taskIds) {
return logRepository.findByExecutionId(tenantId, executionId, minLevel)
.stream()
.filter(logEntry -> levels.contains(logEntry.getLevel().name()));
.filter(data -> taskIds.isEmpty() || taskIds.contains(data.getTaskId()));
}
}
42 changes: 17 additions & 25 deletions webserver/src/main/java/io/kestra/webserver/controllers/api/LogController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,10 @@
import org.slf4j.event.Level;
import reactor.core.publisher.Flux;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.time.ZonedDateTime;
import java.util.List;
import java.util.stream.Collectors;
import java.util.Optional;

import static io.kestra.core.utils.DateUtils.validateTimeline;

Expand Down Expand Up @@ -76,16 +75,14 @@ public List<LogEntry> findByExecution(
@Parameter(description = "The task id") @Nullable @QueryValue String taskId,
@Parameter(description = "The attempt number") @Nullable @QueryValue Integer attempt
) {
if (taskId != null) {
return logRepository.findByExecutionIdAndTaskId(tenantService.resolveTenant(), executionId, taskId, minLevel);
} else if (taskRunId != null) {
if (attempt != null) {
return logRepository.findByExecutionIdAndTaskRunIdAndAttempt(tenantService.resolveTenant(), executionId, taskRunId, minLevel, attempt);
}
return logRepository.findByExecutionIdAndTaskRunId(tenantService.resolveTenant(), executionId, taskRunId, minLevel);
} else {
return logRepository.findByExecutionId(tenantService.resolveTenant(), executionId, minLevel);
}
return logService.getExecutionLogs(
tenantService.resolveTenant(),
executionId,
minLevel,
taskRunId,
Optional.ofNullable(taskId).map(List::of).orElse(null),
attempt
);
}

@ExecuteOn(TaskExecutors.IO)
Expand All @@ -98,19 +95,14 @@ public StreamedFile download(
@Parameter(description = "The task id") @Nullable @QueryValue String taskId,
@Parameter(description = "The attempt number") @Nullable @QueryValue Integer attempt
) {
List<LogEntry> logEntries;
if (taskId != null) {
logEntries = logRepository.findByExecutionIdAndTaskId(tenantService.resolveTenant(), executionId, taskId, minLevel);
} else if (taskRunId != null) {
if (attempt != null) {
logEntries = logRepository.findByExecutionIdAndTaskRunIdAndAttempt(tenantService.resolveTenant(), executionId, taskRunId, minLevel, attempt);
} else {
logEntries = logRepository.findByExecutionIdAndTaskRunId(tenantService.resolveTenant(), executionId, taskRunId, minLevel);
}
} else {
logEntries = logRepository.findByExecutionId(tenantService.resolveTenant(), executionId, minLevel);
}
InputStream inputStream = new ByteArrayInputStream(logEntries.stream().map(LogEntry::toPrettyString).collect(Collectors.joining("\n")).getBytes());
InputStream inputStream = logService.getExecutionLogsAsStream(
tenantService.resolveTenant(),
executionId,
minLevel,
taskRunId,
Optional.ofNullable(taskId).map(List::of).orElse(null),
attempt
);
return new StreamedFile(inputStream, MediaType.TEXT_PLAIN_TYPE).attach(executionId + ".log");
}

Expand Down