feat: add tutorial for lokiExchangeStore

distribution/src/test/java/com/predic8/membrane/tutorials/operation/LokiTutorialTest.java

@@ -0,0 +1,156 @@
+/* Copyright 2026 predic8 GmbH, www.predic8.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 com.predic8.membrane.tutorials.operation;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.predic8.membrane.core.exchange.Exchange;
+import com.predic8.membrane.core.interceptor.AbstractInterceptor;
+import com.predic8.membrane.core.interceptor.Outcome;
+import com.predic8.membrane.core.proxies.ServiceProxy;
+import com.predic8.membrane.core.proxies.ServiceProxyKey;
+import com.predic8.membrane.core.router.DefaultRouter;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static com.predic8.membrane.core.http.Response.ok;
+import static io.restassured.RestAssured.given;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Verifies tutorial step 70-Record-Messages-In-Loki.yaml: lokiExchangeStore pushes every finished exchange to
+ * <code>/loki/api/v1/push</code>, labelled with <code>job</code> and the name of the API.
+ *
+ * <p>A local Membrane router stands in for Loki on the port the tutorial configures, so the
+ * lesson is verified end-to-end without a Loki container. What the tutorial's LogQL steps
+ * promise — a <code>job="membrane-instance-1"</code> stream per API, and lines whose JSON carries method,
+ * URI and status code — is exactly what is asserted here.
+ */
+public class LokiTutorialTest extends AbstractOperationTutorialTest {
+
+    private static final int LOKI_MOCK_PORT = 3100;
+    private static final ObjectMapper om = new ObjectMapper();
+
+    private final List<JsonNode> pushes = new ArrayList<>();
+
+    private DefaultRouter lokiMock;
+
+    @Override
+    protected String getTutorialYaml() {
+        return "70-Record-Messages-In-Loki.yaml";
+    }
+
+    /**
+     * Hides {@code AbstractSampleMembraneStartStopTestcase.startMembrane()} so the Loki stand-in
+     * is listening before Membrane starts and no push can be missed.
+     */
+    @BeforeEach
+    void startMembrane() throws Exception {
+        startLokiMock();
+        process = startServiceProxyScript();
+    }
+
+    @AfterEach
+    void stopLokiMock() {
+        if (lokiMock != null)
+            lokiMock.stop();
+    }
+
+    @Test
+    void exchangesArePushedToLokiLabelledByJobAndApi() throws Exception {
+        // @formatter:off
+        given()
+        .when()
+            .get("http://localhost:2001")
+        .then()
+            .statusCode(200);
+
+        given()
+        .when()
+            .get("http://localhost:2002")
+        .then()
+            .statusCode(404);
+        // @formatter:on
+
+        assertEquals("GET / -> 200", waitForLogLine("Backend A"));
+        assertEquals("GET / -> 404", waitForLogLine("Backend B"));
+    }
+
+    /**
+     * Waits for a log line pushed for the given API and renders it the way the tutorial's
+     * {@code line_format "{{.request_method}} {{.request_uri}} -> {{.response_statusCode}}"}
+     * would.
+     */
+    private String waitForLogLine(String api) throws Exception {
+        // The store batches and pushes every updateIntervalMs (1000 by default).
+        for (int i = 0; i < 100; i++) {
+            JsonNode line = findLogLine(api);
+            if (line != null)
+                return "%s %s -> %d".formatted(
+                        line.get("request").get("method").textValue(),
+                        line.get("request").get("uri").textValue(),
+                        line.get("response").get("statusCode").intValue());
+            Thread.sleep(100);
+        }
+        throw new AssertionError("No exchange pushed to Loki for API '%s' within 10s. Pushes seen: %s"
+                .formatted(api, pushes));
+    }
+
+    private JsonNode findLogLine(String api) throws Exception {
+        synchronized (pushes) {
+            for (JsonNode push : pushes)
+                for (JsonNode stream : push.get("streams")) {
+                    JsonNode labels = stream.get("stream");
+                    if (!"membrane-instance-1".equals(labels.get("job").textValue()))
+                        continue;
+                    if (!api.equals(labels.get("api").textValue()))
+                        continue;
+                    JsonNode values = stream.get("values");
+                    if (!values.isEmpty())
+                        // values[i] is [ "<epoch nanos>", "<exchange as JSON>" ]
+                        return om.readTree(values.get(0).get(1).textValue());
+                }
+        }
+        return null;
+    }
+
+    private void startLokiMock() throws Exception {
+        ServiceProxy sp = new ServiceProxy(new ServiceProxyKey(LOKI_MOCK_PORT), null, 0);
+        sp.getFlow().add(new AbstractInterceptor() {
+            @Override
+            public Outcome handleRequest(Exchange exc) {
+                if (exc.getRequest().getUri().equals("/loki/api/v1/push")) {
+                    try {
+                        synchronized (pushes) {
+                            pushes.add(om.readTree(exc.getRequest().getBodyAsStringDecoded()));
+                        }
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+                exc.setResponse(ok().build());
+                return Outcome.RETURN;
+            }
+        });
+
+        lokiMock = new DefaultRouter();
+        lokiMock.add(sp);
+        lokiMock.start();
+    }
+}

distribution/tutorials/operation/70-Record-Messages-In-Loki.yaml

@@ -0,0 +1,93 @@
+# yaml-language-server: $schema=https://www.membrane-api.io/v7.3.1.json
+#
+# Tutorial: Sending Exchanges to Grafana Loki
+#
+# The prometheus plugin (50-Prometheus.yaml) gives you counters and timings, but it
+# cannot tell you what a single call looked like. lokiExchangeStore fills that gap:
+# it pushes every finished exchange - request and response, headers and bodies - to
+# Grafana Loki as one JSON log line, where you can search it with LogQL.
+#
+# 1.) Start Loki and Grafana:
+#       docker compose -f loki/docker-compose.yml up -d
+#
+# 2.) Start Membrane:
+#       Linux/Mac:  ./membrane.sh -c 70-Record-Messages-In-Loki.yaml
+#       Windows:    membrane.cmd  -c 70-Record-Messages-In-Loki.yaml
+#
+# 3.) Generate traffic on the two APIs:
+#       curl http://localhost:2001
+#       curl http://localhost:2002
+#
+# 4.) Open Grafana at http://localhost:3000 and log in with admin / admin.
+#     Go to Explore (compass icon), pick the Loki datasource, switch the query
+#     editor to "Code" and run:
+#
+#       {job="membrane-instance-1"}
+#
+#     You should see one log line per call, each holding a complete exchange
+#     as JSON.
+#
+# 5.) Unpack that JSON with the 'json' stage and pick out single fields:
+#
+#       {job="membrane-instance-1"} | json | line_format "{{.request_method}} {{.request_uri}} -> {{.response_statusCode}}"
+#
+#     You should see one line per call:
+#
+#       GET / -> 200
+#       GET / -> 404
+#
+# 6.) Every line also carries an 'api' label with the name of the API the
+#     exchange passed through, so you can narrow the search to one API:
+#
+#       {job="membrane-instance-1", api="Backend B"} | json
+#
+#     Labels are indexed by Loki, so filtering on 'api' or 'job' is much cheaper
+#     than filtering on a field inside the JSON.
+#
+# Notes:
+#     Only finished exchanges are pushed. Loki cannot update a log line that has
+#     already been written, so Membrane waits until an exchange is complete and
+#     then writes it once. Exchanges are batched and sent every second; set
+#     updateIntervalMs to change that.
+#
+#     Request and response bodies are byte arrays and therefore Base64-encoded
+#     inside the JSON. Bodies larger than maxBodySize (100000 by default) are
+#     truncated.
+#
+#     The store is write-only: the Membrane admin console cannot browse exchanges
+#     that live in Loki. Use Grafana for that.
+#
+# Troubleshooting:
+#     "While pushing N exchanges to Loki" in the Membrane log means Loki was not
+#     reachable at the configured url. Membrane keeps serving requests and keeps
+#     trying, so once Loki is back up later exchanges arrive. Check the containers:
+#       docker compose -f loki/docker-compose.yml ps
+#
+#     This Grafana also uses port 3000. If you still have the one from
+#     60-Grafana.yaml running, stop it first:
+#       docker compose -f grafana/docker-compose.yml down
+
+components:
+  # A single exchangeStore declared here is picked up automatically: every
+  # exchange passing through the gateway is handed to it.
+  exchangeStore:
+    lokiExchangeStore:
+      url: http://localhost:3100
+      job: membrane-instance-1
+
+---
+# The API name becomes the 'api' label in Loki, so name your APIs.
+api:
+  name: Backend A
+  port: 2001
+  flow:
+    - return:
+        status: 200
+
+---
+api:
+  name: Backend B
+  port: 2002
+  flow:
+    - return:
+        status: 404

distribution/tutorials/operation/loki/docker-compose.yml

@@ -0,0 +1,19 @@
+services:
+  loki:
+    image: grafana/loki:2.9.8
+    container_name: loki
+    ports:
+      - "3100:3100"
+    command: -config.file=/etc/loki/local-config.yaml
+
+  grafana:
+    image: grafana/grafana:9.5.2
+    container_name: grafana-loki
+    ports:
+      - "3000:3000"
+    restart: unless-stopped
+    environment:
+      - GF_SECURITY_ADMIN_USER=admin
+      - GF_SECURITY_ADMIN_PASSWORD=admin
+    volumes:
+      - ./provisioning/datasources:/etc/grafana/provisioning/datasources

distribution/tutorials/operation/loki/provisioning/datasources/datasource.yml

@@ -0,0 +1,8 @@
+apiVersion: 1
+datasources:
+  - name: Loki
+    type: loki
+    uid: P8E80F9AEF21F6940
+    access: proxy
+    url: http://loki:3100
+    isDefault: true