Simple event listener exposing data via HTTP

plugin/trino-http-server-event-listener/pom.xml

@@ -0,0 +1,165 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.trino</groupId>
+        <artifactId>trino-root</artifactId>
+        <version>449-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>trino-http-server-event-listener</artifactId>
+    <packaging>trino-plugin</packaging>
+    <description>Trino event listener exposing collected event via HTTP server</description>
+
+    <properties>
+        <air.main.basedir>${project.parent.basedir}</air.main.basedir>
+        <air.compiler.fail-warnings>true</air.compiler.fail-warnings>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.inject</groupId>
+            <artifactId>guice</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>bootstrap</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>configuration</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>http-server</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>jaxrs</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>json</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>node</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>units</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-cache</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>jakarta.ws.rs</groupId>
+            <artifactId>jakarta.ws.rs-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>slice</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-context</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-spi</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>http-client</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>slice</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.openjdk.jol</groupId>
+                    <artifactId>jol-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>junit-extensions</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-main</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-testing-services</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

plugin/trino-http-server-event-listener/src/main/java/io/trino/plugin/httpquery/HttpServerEventListener.java

@@ -0,0 +1,114 @@
+/*
+ * Licensed 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 io.trino.plugin.httpquery;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Inject;
+import io.airlift.bootstrap.LifeCycleManager;
+import io.airlift.http.server.HttpServerInfo;
+import io.trino.cache.SafeCaches;
+import io.trino.spi.eventlistener.EventListener;
+import io.trino.spi.eventlistener.QueryCompletedEvent;
+import io.trino.spi.eventlistener.QueryCreatedEvent;
+import io.trino.spi.eventlistener.SplitCompletedEvent;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.NotFoundException;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.UriInfo;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.Objects.requireNonNull;
+
+@Path("/v1/events")
+public class HttpServerEventListener
+        implements EventListener
+{
+    private final LifeCycleManager lifecycleManager;
+    private final int serverPort;
+
+    private Cache<String, QueryCompletedEvent> events;
+
+    @Inject
+    public HttpServerEventListener(
+            HttpServerInfo httpServerInfo,
+            LifeCycleManager lifecycleManager,
+            HttpServerEventListenerConfig config)
+    {
+        this.serverPort = httpServerInfo.getHttpUri().getPort();
+        this.lifecycleManager = requireNonNull(lifecycleManager, "lifecycleManager is null");
+        requireNonNull(config, "http event listener config is null");
+        events = SafeCaches.buildNonEvictableCache(CacheBuilder.newBuilder()
+                .maximumSize(config.getEventBufferSize())
+                .expireAfterWrite(config.getEventTTL().toMillis(), TimeUnit.MILLISECONDS));
+    }
+
+    @GET
+    @Path("completedQueries/get/{queryId}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public QueryCompletedEvent get(@PathParam("queryId") String queryId)
+    {
+        QueryCompletedEvent event = events.getIfPresent(queryId);
+        if (event == null) {
+            throw new NotFoundException("completion event for '" + queryId + "' not found");
+        }
+        return event;
+    }
+
+    @GET
+    @Path("completedQueries/list")
+    @Produces(MediaType.APPLICATION_JSON)
+    public List<String> get(@Context UriInfo uriInfo)
+    {
+        return ImmutableList.copyOf(events.asMap().keySet());
+    }
+
+    @Override
+    public void queryCreated(QueryCreatedEvent queryCreatedEvent)
+    {
+        // query creation event not supported
+    }
+
+    @Override
+    public void queryCompleted(QueryCompletedEvent queryCompletedEvent)
+    {
+        events.put(queryCompletedEvent.getMetadata().getQueryId(), queryCompletedEvent);
+    }
+
+    @Override
+    public void splitCompleted(SplitCompletedEvent splitCompletedEvent)
+    {
+        // split completion events not supported
+    }
+
+    @VisibleForTesting
+    int getServerPort()
+    {
+        return serverPort;
+    }
+
+    @Override
+    public void shutdown()
+    {
+        lifecycleManager.stop();
+    }
+}

plugin/trino-http-server-event-listener/src/main/java/io/trino/plugin/httpquery/HttpServerEventListenerConfig.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed 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 io.trino.plugin.httpquery;
+
+import io.airlift.configuration.Config;
+import io.airlift.configuration.ConfigDescription;
+import io.airlift.units.Duration;
+
+import java.util.concurrent.TimeUnit;
+
+public class HttpServerEventListenerConfig
+{
+    private int eventBufferSize = 50;
+    private Duration eventTTL = new Duration(10, TimeUnit.MINUTES);
+
+    @ConfigDescription("Event buffer size")
+    @Config("http-server-event-listener.event-buffer-size")
+    public HttpServerEventListenerConfig setEventBufferSize(int eventBufferSize)
+    {
+        this.eventBufferSize = eventBufferSize;
+        return this;
+    }
+
+    public int getEventBufferSize()
+    {
+        return eventBufferSize;
+    }
+
+    @ConfigDescription("Event TTL")
+    @Config("http-server-event-listener.event-ttl")
+    public HttpServerEventListenerConfig setEventTTL(Duration eventTTL)
+    {
+        this.eventTTL = eventTTL;
+        return this;
+    }
+
+    public Duration getEventTTL()
+    {
+        return eventTTL;
+    }
+}