diff --git a/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpEndpoint.java b/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpEndpoint.java new file mode 100644 index 00000000000..61478109e08 --- /dev/null +++ b/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpEndpoint.java @@ -0,0 +1,56 @@ +/* + * + * 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.bookkeeper.http; + +import java.util.Set; +import lombok.Getter; +import org.apache.bookkeeper.http.service.HttpEndpointService; + +/** + * A value object that binds an HTTP path to its handler service, + * optionally restricted to specific HTTP methods. + */ +@Getter +public class HttpEndpoint { + + private final String path; + private final HttpEndpointService service; + private final Set methods; + + /** + * Create an endpoint that handles all HTTP methods. + */ + public HttpEndpoint(String path, HttpEndpointService service) { + this(path, service, null); + } + + /** + * Create an endpoint restricted to the given HTTP methods. + * + * @param methods the set of allowed methods, or null to allow all methods + */ + public HttpEndpoint(String path, HttpEndpointService service, Set methods) { + this.path = path; + this.service = service; + this.methods = methods; + } + +} diff --git a/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpExtension.java b/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpExtension.java new file mode 100644 index 00000000000..162277f42a4 --- /dev/null +++ b/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpExtension.java @@ -0,0 +1,67 @@ +/* + * + * 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.bookkeeper.http; + +import java.util.List; + +/** + * SPI interface for HTTP endpoint extensions. + * Configured via "httpExtensions" in bk_server.conf. + * + *

One extension class can register multiple endpoints. + * + *

Usage: + *

    + *
  1. Implement this interface
    2. + *
  2. Set httpExtensions=com.example.MyExtension in bk_server.conf
    3. + *
  3. Put the JAR in BookKeeper's classpath
    4. + *
+ * + *

Simple usage (no BK internals needed): + * 

+ * public List<HttpEndpoint> getEndpoints(HttpServiceProvider provider) {
+ *     return Arrays.asList(
+ *         new HttpEndpoint("/api/v1/ext/hello",
+ *             request -> new HttpServiceResponse().setBody("hello"))
+ *     );
+ * }
+ *
+ * + *

Advanced usage (access Bookie internals): + * 

+ * public List<HttpEndpoint> getEndpoints(HttpServiceProvider provider) {
+ *     BKHttpServiceProvider bkProvider = (BKHttpServiceProvider) provider;
+ *     Bookie bookie = bkProvider.getBookieServer().getBookie();
+ *     ...
+ * }
+ *
+ */ +public interface HttpExtension { + + /** + * Return all endpoints to register. + * + * @param provider the HTTP service provider. In BookKeeper, this is + * {@code BKHttpServiceProvider} which provides access to + * {@code BookieServer}, {@code Bookie}, etc. via casting. + */ + List getEndpoints(HttpServiceProvider provider); +} diff --git a/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServer.java b/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServer.java index 71d597d5ffa..607c80a5c99 100644 --- a/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServer.java +++ b/bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServer.java @@ -130,4 +130,12 @@ enum ApiType { * Check whether the HTTP server is still running. */ boolean isRunning(); + + /** + * Set the HTTP extension class names to be loaded by the server. + * + * @param extensionClasses fully-qualified class names of {@link HttpExtension} implementations + */ + default void setHttpExtensionClasses(String[] extensionClasses) { + } } diff --git a/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java b/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java index ecc67debf59..4ffc04a1f31 100644 --- a/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java +++ b/bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java @@ -20,6 +20,7 @@ */ package org.apache.bookkeeper.http.vertx; +import com.google.common.base.Strings; import io.vertx.core.AbstractVerticle; import io.vertx.core.AsyncResult; import io.vertx.core.Vertx; @@ -27,12 +28,18 @@ import io.vertx.core.http.HttpServerOptions; import io.vertx.core.net.JksOptions; import io.vertx.ext.web.Router; +import io.vertx.ext.web.RoutingContext; import io.vertx.ext.web.handler.BodyHandler; import java.io.IOException; +import java.util.EnumSet; +import java.util.List; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import lombok.CustomLog; +import org.apache.bookkeeper.http.HttpEndpoint; +import org.apache.bookkeeper.http.HttpExtension; import org.apache.bookkeeper.http.HttpRouter; import org.apache.bookkeeper.http.HttpServer; import org.apache.bookkeeper.http.HttpServerConfiguration; @@ -47,6 +54,7 @@ public class VertxHttpServer implements HttpServer { private final Vertx vertx; private boolean isRunning; private HttpServiceProvider httpServiceProvider; + private String[] httpExtensionClasses; private int listeningPort = -1; public VertxHttpServer() { @@ -62,6 +70,14 @@ public void initialize(HttpServiceProvider httpServiceProvider) { this.httpServiceProvider = httpServiceProvider; } + /** + * Set the HTTP extension class names to load. + */ + @Override + public void setHttpExtensionClasses(String[] extensionClasses) { + this.httpExtensionClasses = extensionClasses; + } + @Override public boolean startServer(int port) { return startServer(port, "0.0.0.0"); @@ -88,6 +104,7 @@ public void bindHandler(String endpoint, VertxAbstractHandler handler) { } }; requestRouter.bindAll(); + registerExtensions(router); vertx.deployVerticle(new AbstractVerticle() { @Override public void start() throws Exception { @@ -130,6 +147,59 @@ public void start() throws Exception { return false; } + /** + * Load and register all configured HTTP extensions. + */ + private void registerExtensions(Router router) { + if (httpExtensionClasses == null || httpExtensionClasses.length == 0) { + return; + } + for (String className : httpExtensionClasses) { + if (Strings.isNullOrEmpty(className)) { + continue; + } + try { + Class cls = + Class.forName(className.trim()).asSubclass(HttpExtension.class); + HttpExtension ext = cls.getDeclaredConstructor().newInstance(); + List endpoints = ext.getEndpoints(httpServiceProvider); + for (HttpEndpoint endpoint : endpoints) { + String path = endpoint.getPath(); + if (path == null || !path.startsWith("/")) { + log.warn().attr("path", path).log( + "Skipping invalid extension path (must be non-null and start with '/')"); + continue; + } + log.info().attr("path", path).attr("class", className).log("Loading HTTP extension"); + VertxAbstractHandler handler = new VertxAbstractHandler() { + @Override + public void handle(RoutingContext ctx) { + processRequest(endpoint.getService(), ctx); + } + }; + Set methods = endpoint.getMethods(); + if (methods == null) { + methods = EnumSet.allOf(HttpServer.Method.class); + } + if (methods.contains(HttpServer.Method.GET)) { + router.get(path).blockingHandler(handler); + } + if (methods.contains(HttpServer.Method.PUT)) { + router.put(path).blockingHandler(handler); + } + if (methods.contains(HttpServer.Method.POST)) { + router.post(path).blockingHandler(handler); + } + if (methods.contains(HttpServer.Method.DELETE)) { + router.delete(path).blockingHandler(handler); + } + } + } catch (ReflectiveOperationException e) { + log.error().exception(e).attr("class", className).log("Failed to load HTTP extension"); + } + } + } + @Override public void stopServer() { CountDownLatch shutdownLatch = new CountDownLatch(1); diff --git a/bookkeeper-http/vertx-http-server/src/test/java/org/apache/bookkeeper/http/vertx/TestVertxHttpServerExtension.java b/bookkeeper-http/vertx-http-server/src/test/java/org/apache/bookkeeper/http/vertx/TestVertxHttpServerExtension.java new file mode 100644 index 00000000000..39824f54d11 --- /dev/null +++ b/bookkeeper-http/vertx-http-server/src/test/java/org/apache/bookkeeper/http/vertx/TestVertxHttpServerExtension.java @@ -0,0 +1,283 @@ +/* + * + * 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.bookkeeper.http.vertx; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumSet; +import java.util.List; +import org.apache.bookkeeper.http.HttpEndpoint; +import org.apache.bookkeeper.http.HttpExtension; +import org.apache.bookkeeper.http.HttpRouter; +import org.apache.bookkeeper.http.HttpServer; +import org.apache.bookkeeper.http.HttpServiceProvider; +import org.apache.bookkeeper.http.NullHttpServiceProvider; +import org.apache.bookkeeper.http.service.HttpEndpointService; +import org.apache.bookkeeper.http.service.HttpServiceRequest; +import org.apache.bookkeeper.http.service.HttpServiceResponse; +import org.junit.Test; + +/** + * Unit test for HTTP extension SPI loading and routing in {@link VertxHttpServer}. + */ +public class TestVertxHttpServerExtension { + + /** + * A test extension that registers a GET-only hello endpoint. + */ + public static class GetOnlyExtension implements HttpExtension { + @Override + public List getEndpoints(HttpServiceProvider provider) { + return Collections.singletonList( + new HttpEndpoint("/api/v1/ext/hello", + new HelloService(), + EnumSet.of(HttpServer.Method.GET))); + } + } + + /** + * A test extension that registers an endpoint accepting all HTTP methods. + */ + public static class AllMethodsExtension implements HttpExtension { + @Override + public List getEndpoints(HttpServiceProvider provider) { + return Collections.singletonList( + new HttpEndpoint("/api/v1/ext/all", new HelloService())); + } + } + + /** + * A test extension that registers multiple endpoints. + */ + public static class MultiEndpointExtension implements HttpExtension { + @Override + public List getEndpoints(HttpServiceProvider provider) { + return Arrays.asList( + new HttpEndpoint("/api/v1/ext/multi/hello", new HelloService()), + new HttpEndpoint("/api/v1/ext/multi/echo", + new EchoService(), + EnumSet.of(HttpServer.Method.POST))); + } + } + + /** + * An extension that returns an endpoint with an invalid (non-absolute) path. + */ + public static class InvalidPathExtension implements HttpExtension { + @Override + public List getEndpoints(HttpServiceProvider provider) { + return Collections.singletonList( + new HttpEndpoint("no-leading-slash", new HelloService())); + } + } + + static class HelloService implements HttpEndpointService { + @Override + public HttpServiceResponse handle(HttpServiceRequest request) { + return new HttpServiceResponse("hello", HttpServer.StatusCode.OK); + } + } + + static class EchoService implements HttpEndpointService { + @Override + public HttpServiceResponse handle(HttpServiceRequest request) { + return new HttpServiceResponse(request.getBody(), HttpServer.StatusCode.OK); + } + } + + @Test + public void testExtensionEndpoint_GET() throws Exception { + VertxHttpServer httpServer = new VertxHttpServer(); + HttpServiceProvider httpServiceProvider = NullHttpServiceProvider.getInstance(); + httpServer.initialize(httpServiceProvider); + httpServer.setHttpExtensionClasses( + new String[]{GetOnlyExtension.class.getName()}); + assertTrue(httpServer.startServer(0)); + int port = httpServer.getListeningPort(); + + HttpResponse httpResponse = send(getUrl(port, "/api/v1/ext/hello"), HttpServer.Method.GET); + assertEquals(HttpServer.StatusCode.OK.getValue(), httpResponse.responseCode); + assertEquals("hello", httpResponse.responseBody); + httpServer.stopServer(); + } + + @Test + public void testExtensionEndpoint_MethodRestricted() throws Exception { + VertxHttpServer httpServer = new VertxHttpServer(); + HttpServiceProvider httpServiceProvider = NullHttpServiceProvider.getInstance(); + httpServer.initialize(httpServiceProvider); + httpServer.setHttpExtensionClasses( + new String[]{GetOnlyExtension.class.getName()}); + assertTrue(httpServer.startServer(0)); + int port = httpServer.getListeningPort(); + + // POST should return 405 since only GET is allowed + HttpResponse httpResponse = send(getUrl(port, "/api/v1/ext/hello"), HttpServer.Method.POST); + assertEquals(HttpServer.StatusCode.METHOD_NOT_ALLOWED.getValue(), httpResponse.responseCode); + httpServer.stopServer(); + } + + @Test + public void testExtensionEndpoint_AllMethods() throws Exception { + VertxHttpServer httpServer = new VertxHttpServer(); + HttpServiceProvider httpServiceProvider = NullHttpServiceProvider.getInstance(); + httpServer.initialize(httpServiceProvider); + httpServer.setHttpExtensionClasses( + new String[]{AllMethodsExtension.class.getName()}); + assertTrue(httpServer.startServer(0)); + int port = httpServer.getListeningPort(); + + for (HttpServer.Method method : HttpServer.Method.values()) { + HttpResponse httpResponse = send(getUrl(port, "/api/v1/ext/all"), method); + assertEquals("Method " + method + " should return 200", + HttpServer.StatusCode.OK.getValue(), httpResponse.responseCode); + } + httpServer.stopServer(); + } + + @Test + public void testExtensionEndpoint_InvalidPathSkipped() throws Exception { + VertxHttpServer httpServer = new VertxHttpServer(); + HttpServiceProvider httpServiceProvider = NullHttpServiceProvider.getInstance(); + httpServer.initialize(httpServiceProvider); + httpServer.setHttpExtensionClasses( + new String[]{InvalidPathExtension.class.getName()}); + assertTrue(httpServer.startServer(0)); + int port = httpServer.getListeningPort(); + + // The invalid path should be skipped, so requesting it returns 404 + HttpResponse httpResponse = send(getUrl(port, "/no-leading-slash"), HttpServer.Method.GET); + assertEquals(HttpServer.StatusCode.NOT_FOUND.getValue(), httpResponse.responseCode); + httpServer.stopServer(); + } + + @Test + public void testExtensionClassNotFound_Graceful() throws Exception { + VertxHttpServer httpServer = new VertxHttpServer(); + HttpServiceProvider httpServiceProvider = NullHttpServiceProvider.getInstance(); + httpServer.initialize(httpServiceProvider); + httpServer.setHttpExtensionClasses( + new String[]{"com.nonexistent.ExtensionClass"}); + assertTrue(httpServer.startServer(0)); + int port = httpServer.getListeningPort(); + + // Built-in endpoints should still work + HttpResponse httpResponse = send(getUrl(port, HttpRouter.HEARTBEAT), HttpServer.Method.GET); + assertEquals(HttpServer.StatusCode.OK.getValue(), httpResponse.responseCode); + httpServer.stopServer(); + } + + @Test + public void testNoExtensions_Configured() throws Exception { + VertxHttpServer httpServer = new VertxHttpServer(); + HttpServiceProvider httpServiceProvider = NullHttpServiceProvider.getInstance(); + httpServer.initialize(httpServiceProvider); + // No extension classes configured + assertTrue(httpServer.startServer(0)); + int port = httpServer.getListeningPort(); + + HttpResponse httpResponse = send(getUrl(port, HttpRouter.HEARTBEAT), HttpServer.Method.GET); + assertEquals(HttpServer.StatusCode.OK.getValue(), httpResponse.responseCode); + httpServer.stopServer(); + } + + @Test + public void testMultipleExtensions() throws Exception { + VertxHttpServer httpServer = new VertxHttpServer(); + HttpServiceProvider httpServiceProvider = NullHttpServiceProvider.getInstance(); + httpServer.initialize(httpServiceProvider); + httpServer.setHttpExtensionClasses( + new String[]{MultiEndpointExtension.class.getName()}); + assertTrue(httpServer.startServer(0)); + int port = httpServer.getListeningPort(); + + // First endpoint (GET) + HttpResponse helloResp = send(getUrl(port, "/api/v1/ext/multi/hello"), HttpServer.Method.GET); + assertEquals(HttpServer.StatusCode.OK.getValue(), helloResp.responseCode); + assertEquals("hello", helloResp.responseBody); + + // Second endpoint (POST) + String body = "echo-test"; + HttpResponse echoResp = send(getUrl(port, "/api/v1/ext/multi/echo"), HttpServer.Method.POST, body); + assertEquals(HttpServer.StatusCode.OK.getValue(), echoResp.responseCode); + assertEquals(body, echoResp.responseBody); + httpServer.stopServer(); + } + + // --- helper methods --- + + private HttpResponse send(String url, HttpServer.Method method) throws IOException { + return send(url, method, ""); + } + + private HttpResponse send(String url, HttpServer.Method method, String body) throws IOException { + URL obj = new URL(url); + HttpURLConnection con = (HttpURLConnection) obj.openConnection(); + con.setRequestMethod(method.toString()); + if (!body.isEmpty()) { + con.setDoOutput(true); + con.setFixedLengthStreamingMode(body.length()); + con.getOutputStream().write(body.getBytes(StandardCharsets.UTF_8)); + con.getOutputStream().flush(); + } + int responseCode = con.getResponseCode(); + StringBuilder response = new StringBuilder(); + java.io.InputStream stream = responseCode >= 400 ? con.getErrorStream() : con.getInputStream(); + BufferedReader in = null; + try { + if (stream != null) { + in = new BufferedReader(new InputStreamReader(stream)); + String inputLine; + while ((inputLine = in.readLine()) != null) { + response.append(inputLine); + } + } + } finally { + if (in != null) { + in.close(); + } + } + return new HttpResponse(responseCode, response.toString()); + } + + private String getUrl(int port, String path) { + return "http://localhost:" + port + path; + } + + private static class HttpResponse { + private final int responseCode; + private final String responseBody; + + HttpResponse(int responseCode, String responseBody) { + this.responseCode = responseCode; + this.responseBody = responseBody; + } + } +} diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java index a6ce37874db..0ccfc5a2f9d 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java @@ -285,6 +285,9 @@ public class ServerConfiguration extends AbstractConfiguration