[observability] Merge observability into http-server

fusion-documentation/pom.xml

@@ -64,7 +64,7 @@
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
-      <artifactId>fusion-observability</artifactId>
+        <artifactId>fusion-http-server</artifactId>
       <version>${project.version}</version>
       <scope>provided</scope>
     </dependency>

fusion-http-server/pom.xml

@@ -28,6 +28,12 @@
   <name>Fusion :: Http Server</name>
 
   <dependencies>
+      <dependency>
+          <groupId>${project.groupId}</groupId>
+          <artifactId>fusion-build-api</artifactId>
+          <version>${project.version}</version>
+          <scope>provided</scope>
+      </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>fusion-api</artifactId>

fusion-http-server/src/main/java/io/yupiik/fusion/http/server/impl/health/Health.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright (c) 2022 - present - Yupiik SAS - https://www.yupiik.com
+ * 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.yupiik.fusion.http.server.impl.health;
+
+import io.yupiik.fusion.http.server.api.Request;
+import io.yupiik.fusion.http.server.api.Response;
+import io.yupiik.fusion.http.server.spi.MonitoringEndpoint;
+import io.yupiik.fusion.http.server.impl.health.HealthCheck;
+import io.yupiik.fusion.http.server.impl.health.HealthRegistry;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CompletionStage;
+
+import static java.util.Optional.ofNullable;
+import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.function.Function.identity;
+import static java.util.stream.Collectors.joining;
+import static java.util.stream.Collectors.toMap;
+
+public class Health implements MonitoringEndpoint {
+    private final HealthRegistry healthChecks;
+
+    public Health(final HealthRegistry healthChecks) {
+        this.healthChecks = healthChecks;
+    }
+
+    @Override
+    public boolean matches(final Request request) {
+        return "GET".equalsIgnoreCase(request.method()) && "/health".equalsIgnoreCase(request.path());
+    }
+
+    @Override
+    public CompletionStage<Response> handle(final Request request) {
+        final var query = request.query();
+        final var filter = query != null && query.startsWith("type=") ?
+                query.substring("type=".length()) : null;
+
+        var healthCheckStream = healthChecks.healthChecks().stream();
+        if (filter != null) {
+            healthCheckStream = healthCheckStream.filter(c -> Objects.equals(c.type(), filter));
+        }
+        final var checks = healthCheckStream
+                .collect(toMap(identity(), c -> c.check().toCompletableFuture()));
+        return allOf(checks.values().toArray(new CompletableFuture<?>[0]))
+                .thenApply(success -> success(checks, request))
+                .exceptionally(failed -> failure(checks));
+    }
+
+    private Response failure(final Map<HealthCheck, CompletableFuture<HealthCheck.Result>> checks) {
+        final var response = Response.of()
+                .status(503)
+                .header("content-type", "text/plain")
+                .body(checks.entrySet().stream()
+                        .map(c -> {
+                            try {
+                                return toSuccessLine(c);
+                            } catch (final CompletionException | CancellationException ce) {
+                                return c.getKey().name() + ",KO,\"" + ofNullable(ce.getMessage()).map(this::escape).orElse("") + "\"";
+                            }
+                        })
+                        .collect(joining("\n")))
+                .build();
+
+        // cancel if any is still pending
+        checks.values().stream()
+                .filter(it -> !it.isCompletedExceptionally())
+                .forEach(it -> {
+                    try {
+                        it.cancel(true);
+                    } catch (final RuntimeException re) {
+                        // no-op
+                    }
+                });
+        return response;
+    }
+
+    private Response success(final Map<HealthCheck, CompletableFuture<HealthCheck.Result>> checks, final Request request) {
+        final var hasFailure = checks.values().stream()
+                .anyMatch(it -> it.getNow(null).status() == HealthCheck.Status.KO);
+        if (!hasFailure) request.setAttribute("skip-access-log", true);
+        return Response.of()
+                .status(hasFailure ? 503 : 200)
+                .header("content-type", "text/plain")
+                .body(checks.entrySet().stream()
+                        .map(this::toSuccessLine)
+                        .collect(joining("\n")))
+                .build();
+    }
+
+    private String toSuccessLine(final Map.Entry<HealthCheck, CompletableFuture<HealthCheck.Result>> c) {
+        final var result = c.getValue().getNow(null);
+        return c.getKey().name() + "," + result.status() + (result.message() != null ? ",\"" + escape(result.message()) + '"' : "");
+    }
+
+    private String escape(final String string) {
+        return string.replace("\"", "\\\"").replace("\n", "\\n");
+    }
+}

fusion-http-server/src/main/java/io/yupiik/fusion/http/server/impl/health/HealthCheck.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2022 - present - Yupiik SAS - https://www.yupiik.com
+ * 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.yupiik.fusion.http.server.impl.health;
+
+import java.util.concurrent.CompletionStage;
+
+public interface HealthCheck {
+    /**
+     * @return identifier for this healthcheck.
+     */
+    String name();
+
+    /**
+     * @return type to match when calling {@code /health?type=xxxx} endpoint.
+     */
+    default String type() {
+        return "live";
+    }
+
+    /**
+     * IMPORTANT: ensure to implement some consistent timeouts and error handling if you deploy in kubernetes to avoid to hang any healthcheck or require fusion to cancel them.
+     *
+     * @return the implementation of the check itself, key point is to return the status in the result.
+     */
+    CompletionStage<Result> check();
+
+    record Result(Status status, String message) {
+    }
+
+    enum Status {
+        OK, KO
+    }
+}

fusion-http-server/src/main/java/io/yupiik/fusion/http/server/impl/health/HealthRegistry.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2022 - present - Yupiik SAS - https://www.yupiik.com
+ * 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.yupiik.fusion.http.server.impl.health;
+
+import io.yupiik.fusion.framework.api.scope.ApplicationScoped;
+import io.yupiik.fusion.http.server.impl.health.HealthCheck;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+@ApplicationScoped
+public class HealthRegistry {
+    private final List<HealthCheck> healthChecks;
+
+    protected HealthRegistry() {
+        this(null);
+    }
+
+    public HealthRegistry(final List<HealthCheck> healthChecks) {
+        this.healthChecks = healthChecks == null ? null : new CopyOnWriteArrayList<>(healthChecks);
+    }
+
+    public Removable register(final HealthCheck check) {
+        healthChecks.add(check);
+        return () -> healthChecks.remove(check);
+    }
+
+    public List<HealthCheck> healthChecks() {
+        return healthChecks;
+    }
+
+    public interface Removable extends AutoCloseable {
+        @Override
+        void close();
+    }
+}

fusion-http-server/src/main/java/io/yupiik/fusion/http/server/impl/http/MonitoringEndpointRegistry.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) 2022 - present - Yupiik SAS - https://www.yupiik.com
+ * 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.yupiik.fusion.http.server.impl.http;
+
+import io.yupiik.fusion.framework.api.scope.DefaultScoped;
+import io.yupiik.fusion.http.server.spi.MonitoringEndpoint;
+import io.yupiik.fusion.http.server.impl.health.Health;
+import io.yupiik.fusion.http.server.impl.health.HealthRegistry;
+import io.yupiik.fusion.http.server.impl.metrics.Metrics;
+import io.yupiik.fusion.http.server.impl.metrics.MetricsRegistry;
+
+import java.util.List;
+
+@DefaultScoped
+public class MonitoringEndpointRegistry {
+    private final List<MonitoringEndpoint> endpoints;
+
+    public MonitoringEndpointRegistry(final HealthRegistry health, final MetricsRegistry metrics) {
+        this.endpoints = List.of(new Health(health), new Metrics(metrics));
+    }
+
+    public List<MonitoringEndpoint> endpoints() {
+        return endpoints;
+    }
+}

fusion-http-server/src/main/java/io/yupiik/fusion/http/server/impl/metrics/Metrics.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2022 - present - Yupiik SAS - https://www.yupiik.com
+ * 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.yupiik.fusion.http.server.impl.metrics;
+
+import io.yupiik.fusion.http.server.api.Request;
+import io.yupiik.fusion.http.server.api.Response;
+import io.yupiik.fusion.http.server.spi.MonitoringEndpoint;
+
+import java.util.concurrent.CompletionStage;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+
+public class Metrics implements MonitoringEndpoint {
+    private final MetricsRegistry registry;
+    private final OpenMetricsFormatter formatter = new OpenMetricsFormatter();
+
+    public Metrics(final MetricsRegistry registry) {
+        this.registry = registry;
+    }
+
+    @Override
+    public boolean matches(final Request request) {
+        return "GET".equalsIgnoreCase(request.method()) && "/metrics".equalsIgnoreCase(request.path());
+    }
+
+    @Override
+    public CompletionStage<Response> handle(final Request request) {
+        request.setAttribute("skip-access-log", true);
+        return completedFuture(Response.of()
+                .status(200)
+                .header("content-type", "text/plain")
+                .body(formatter.apply(registry.entries()))
+                .build());
+    }
+}