afs commented on code in PR #3184: URL: https://github.com/apache/jena/pull/3184#discussion_r2303495798
########## jena-fuseki2/jena-fuseki-mod-exectracker/src/main/java/org/apache/jena/fuseki/mod/exec/tracker/ExecTrackerService.java: ########## @@ -0,0 +1,397 @@ +/* + * 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.jena.fuseki.mod.exec.tracker; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.OutputStreamWriter; +import java.io.PrintWriter; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.commons.io.IOUtils; +import org.apache.jena.fuseki.FusekiException; +import org.apache.jena.fuseki.server.Endpoint; +import org.apache.jena.fuseki.servlets.BaseActionREST; +import org.apache.jena.fuseki.servlets.HttpAction; +import org.apache.jena.riot.WebContent; +import org.apache.jena.sparql.exec.tracker.BasicTaskExec; +import org.apache.jena.sparql.exec.tracker.TaskEventHistory; +import org.apache.jena.sparql.exec.tracker.TaskListener; +import org.apache.jena.sparql.util.Context; +import org.apache.jena.web.HttpSC; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.gson.Gson; +import com.google.gson.JsonElement; +import com.google.gson.internal.bind.JsonTreeWriter; +import com.google.gson.stream.JsonWriter; + +import jakarta.servlet.AsyncContext; +import jakarta.servlet.AsyncEvent; +import jakarta.servlet.AsyncListener; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; + +/** + * REST action handler for listing running query executions and stopping them. + */ +public class ExecTrackerService extends BaseActionREST { + private static final Logger logger = LoggerFactory.getLogger(ExecTrackerService.class); + + // Gson for formatting server side event (SSE) JSON. + // Note: Pretty printing breaks SSE events due to newlines! + private static Gson gsonForSseEvents = new Gson(); + + /** Helper class to track SSE clients. */ + private class Clients { + // Lock to prevent concurrent addition/removal of listeners while broadcasting events. + Object listenerLock = new Object(); + + // The endpoint of the clients. + Endpoint endpoint; + + // Single listener on an TaskTrackerRegistry. - Not needed here; this listener initialized during FMOD init. + Runnable taskTrackerListenerDisposer; + + // Web clients on the ExecTracker. + Map<AsyncContext, Runnable> eventListeners = Collections.synchronizedMap(new IdentityHashMap<>()); // new ConcurrentHashMap<>(); + + // The history tracker connected to the taskTracker. + // TaskEventHistory historyTracker; + } + + /** Registered clients listening to server side events for indexer status updates. */ + private Map<TaskEventHistory, Clients> trackerToClients = new ConcurrentHashMap<>(); // Collections.synchronizedMap(new IdentityHashMap<>()); + + public ExecTrackerService() {} + + private static long getExecId(HttpAction action) { + String str = action.getRequest().getParameter("requestId"); + Objects.requireNonNull(str); + long result = Long.parseLong(str); + return result; + } + + /** + * The GET command can serve: the website, the notification stream from task execution + * and the latest task execution status. + */ + @Override + protected void doGet(HttpAction action) { + String rawCommand = action.getRequestParameter("command"); + String command = Optional.ofNullable(rawCommand).orElse("page"); + switch (command) { + case "page": servePage(action); break; + case "events": serveEvents(action); break; + case "status": serveStatus(action); break; + case "stop": stopExec(action); break; + default: + throw new UnsupportedOperationException("Unsupported operation: " + command); + } + } + + protected void stopExec(HttpAction action) { + checkIsAbortAllowed(action); + + long execId = getExecId(action); + + TaskEventHistory taskEventHistory = requireTaskEventHistory(action); + BasicTaskExec task = taskEventHistory.getTaskBySerialId(execId); + + if (task != null) { + try { + task.abort(); + } catch (Throwable t) { + logger.warn("Exception raised during abort.", t); + } + logger.info("Sending stop request to execution: " + execId); + } else { + logger.warn("No such execution to abort: " + execId); + } + + respond(action, HttpSC.OK_200, WebContent.contentTypeTextPlain, "Abort request accepted."); + } + + protected void servePage(HttpAction action) { + // Serves the minimal graphql ui + String resourceName = "exec-tracker/index.html"; + String str = null; + try (InputStream in = ExecTrackerService.class.getClassLoader().getResourceAsStream(resourceName)) { + str = IOUtils.toString(in, StandardCharsets.UTF_8); + } catch (IOException e) { + throw new FusekiException(e); + } + + if (str == null) { + respond(action, HttpSC.INTERNAL_SERVER_ERROR_500, WebContent.contentTypeTextPlain, + "Failed to load classpath resource " + resourceName); + } else { + respond(action, HttpSC.OK_200, WebContent.contentTypeHTML, str); + } + } + + protected TaskEventHistory requireTaskEventHistory(HttpAction action) { + Context cxt = action.getEndpoint().getContext(); + TaskEventHistory taskEventHistory = TaskEventHistory.require(cxt); + return taskEventHistory; + } + + protected Runnable registerTaskEventListener(TaskEventHistory taskEventHistory, Clients clients) { + // Register the SSE handler to the history tracker + InternalListener listener = new InternalListener(taskEventHistory, clients); + Runnable disposeTaskEventListener = taskEventHistory.addListener(BasicTaskExec.class, listener); + return disposeTaskEventListener; + } + + protected class InternalListener implements TaskListener<BasicTaskExec> { + protected TaskEventHistory taskEvenHistory; + protected Clients clients; + + Long getTaskId(BasicTaskExec task) { + long taskId = taskEvenHistory.getId(task); + Long serialId = taskEvenHistory.getSerialId(taskId); + return serialId; + } + + public InternalListener(TaskEventHistory taskEventHistory, Clients clients) { + super(); + this.taskEvenHistory = taskEventHistory; + this.clients = clients; + } + + @Override + public void onStateChange(BasicTaskExec task) { + switch (task.getTaskState()) { + case STARTING: onStart(task); break; + case TERMINATED: onTerminated(task); break; + default: // ignored + } + } + + public void onStart(BasicTaskExec startRecord) { + Long serialId = getTaskId(startRecord); + if (serialId != null) { + try (JsonTreeWriter writer = new JsonTreeWriter()) { + writer.beginObject(); + TaskStatusWriter.writeStartRecordMembers(writer, serialId, startRecord); + TaskStatusWriter.writeCanAbort(writer, isAbortAllowed(clients.endpoint)); + writer.endObject(); + JsonElement json = writer.get(); + synchronized (clients.listenerLock) { + Iterator<Entry<AsyncContext, Runnable>> it = clients.eventListeners.entrySet().iterator(); + broadcastJson(it, json); + } + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + public void onTerminated(BasicTaskExec endRecord) { + Long serialId = getTaskId(endRecord); + if (serialId != null) { + try (JsonTreeWriter writer = new JsonTreeWriter()) { + TaskStatusWriter.writeCompletionRecordObject(writer, serialId, endRecord); + JsonElement json = writer.get(); + synchronized (clients.listenerLock) { + Iterator<Entry<AsyncContext, Runnable>> it = clients.eventListeners.entrySet().iterator(); + broadcastJson(it, json); + } + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + } + protected void serveEvents(HttpAction action) { Review Comment: This ("?command=events")is showing up as (-1, -1) in the log. Maybe this will help: ``` action.setResponseStatus(HttpSC.OK_200); ``` It would also benefit from a `try-catch` just in case an internal error happens. ########## jena-fuseki2/jena-fuseki-mod-exectracker/src/main/resources/exec-tracker/index.html: ########## @@ -0,0 +1,477 @@ +<!DOCTYPE html> +<html lang="en"> +<head> + <meta charset="UTF-8" /> + <meta name="viewport" content="width=device-width, initial-scale=1.0" /> + <title>Execution Tracker</title> + + <style> + .controls { + display: flex; + box-sizing: border-box; + width: 100%; + gap: 10px; + margin-bottom: 20px; + align-items: center; + background: #f8f9fa; + padding: 10px; + border-radius: 5px; + } + + .log-info { + background-color: #cce5ff; + color: #004085; + } + + table { + width: 100%; + border-collapse: collapse; + margin-bottom: 2em; + } + th, td { + padding: 0.5em; + border: 1px solid #ccc; + text-align: left; + overflow-wrap: break-word; + vertical-align: top; + } + th:nth-child(1), td:nth-child(1) { + width: 8em; + white-space: nowrap; + } + th:nth-child(2), td:nth-child(2), + th:nth-child(3), td:nth-child(3) { + width: 12em; + white-space: nowrap; + } + th:nth-child(4), td:nth-child(4) { + width: auto; + } + .expandable { + cursor: pointer; + color: blue; + text-decoration: underline; + display: block; + } + .error { + background-color: #fdd; + } + .details-box-row { + display: none; + background-color: #f5f5f5; + } + .details-box { + padding: 0.5em; + font-family: monospace; + text-align: left; + word-wrap: break-word; + white-space: pre-wrap; + /* border-top: 1px solid #ccc; */ + } + .fade-out { + opacity: 0.2; + transition: opacity 10s ease; + } + .fade-in { + opacity: 0.2; + transition: opacity 3s ease; + } + .fade-in.show { + opacity: 1; + } + </style> +</head> +<body> + + <!-- Banner that indicates manual refresh when there is too much activity. --> + <div id="reload-banner" class="controls log-info" style="flex-direction: column; display: none;"> + <div id="status-container" style="width: 100%;"> + <pre id="status-message">Too much activity. Please refresh manually.</pre> + </div> + </div> + + <h1>Ongoing Executions</h1> + <table id="runningTasksTable"> + <thead> + <tr><th>ID</th><th>Start Time</th><th>Label</th><th>Action</th></tr> Review Comment: This is not a "Label". I am seeing the whole query and I'm using short queries. Is there a way to set the display? Maybe - strip prefixes, use an abbreviated string of the query. My first reaction on seeing the query string was that clicking would run the query. Saying "See Details" or some such would be clearer. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
