Prometheus labels API: add response listener

x-pack/plugin/prometheus/src/main/java/org/elasticsearch/xpack/prometheus/rest/PrometheusLabelsResponseListener.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.prometheus.rest;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
+import org.elasticsearch.rest.RestChannel;
+import org.elasticsearch.rest.RestResponse;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.json.JsonXContent;
+import org.elasticsearch.xpack.esql.action.EsqlQueryResponse;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Converts an {@link EsqlQueryResponse} from a Prometheus labels ESQL plan into the
+ * Prometheus {@code /api/v1/labels} JSON response format.
+ *
+ * <p>After the plan executes, the response has a single {@code dimension_fields} column with one
+ * label name per row (already deduplicated and sorted by the ESQL plan). Values have the
+ * {@code "labels."} prefix stripped before being returned. OTel dimension fields carry no such
+ * prefix (e.g. {@code "attributes.service.name"}) and are returned as-is.
+ *
+ * <p>{@code __name__} is always emitted as the first label in the response. For Prometheus metrics
+ * it is present in {@code dimension_fields} as {@code "labels.__name__"} and is deduplicated by
+ * the serialisation. For OTel metrics it is absent from {@code dimension_fields} but is injected
+ * here to signal to clients (e.g. Grafana) that the metric-name label is always available.
+ */
+public class PrometheusLabelsResponseListener {
+
+    private static final Logger logger = LogManager.getLogger(PrometheusLabelsResponseListener.class);
+
+    private static final String LABELS_PREFIX = "labels.";
+    private static final String NAME_LABEL = "__name__";
+    private static final String CONTENT_TYPE = "application/json";
+
+    private PrometheusLabelsResponseListener() {}
+
+    static ActionListener<EsqlQueryResponse> create(RestChannel channel, int limit) {
+        return ActionListener.wrap(response -> {
+            List<String> labelNames = collectLabelNames(response.rows());
+            channel.sendResponse(buildSuccessResponse(labelNames, limit));
+        }, e -> {
+            logger.debug("Labels request failed", e);
+            PrometheusErrorResponse.send(channel, e, logger);
+        });
+    }
+
+    /**
+     * Collects label names from the single-column ESQL result rows, stripping the
+     * {@code "labels."} prefix. OTel dimension fields (e.g. {@code "attributes.service.name"})
+     * carry no such prefix and are returned as-is. Package-private for testing.
+     */
+    static List<String> collectLabelNames(Iterable<? extends Iterable<Object>> rows) {
+        List<String> result = new ArrayList<>();
+        for (Iterable<Object> row : rows) {
+            Object value = row.iterator().next();
+            if (value == null) {
+                continue;
+            }
+            String raw = value.toString();
+            String labelName = raw.startsWith(LABELS_PREFIX) ? raw.substring(LABELS_PREFIX.length()) : raw;
+            if (labelName.isEmpty() == false) {
+                result.add(labelName);
+            }
+        }
+        return result;
+    }
+
+    /**
+     * Builds the success response. {@code __name__} is always written first (synthetic for OTel,
+     * deduped for Prometheus). The {@code limit} heuristic: if the ESQL result contains exactly
+     * {@code limit} entries the response may have been truncated, so a {@code warnings} entry is
+     * added. Package-private for testing.
+     */
+    static RestResponse buildSuccessResponse(List<String> labelNames, int limit) throws IOException {
+        boolean truncated = limit > 0 && labelNames.size() == limit;
+        XContentBuilder builder = JsonXContent.contentBuilder();
+        builder.startObject();
+        builder.field("status", "success");
+        builder.startArray("data");
+        // __name__ is always first — it sorts before 'a' in ASCII so this is also correct order
+        builder.value(NAME_LABEL);
+        for (String name : labelNames) {
+            if (NAME_LABEL.equals(name) == false) {
+                builder.value(name);
+            }
+        }
+        builder.endArray();
+        if (truncated) {
+            builder.startArray("warnings");
+            builder.value("results truncated due to limit");
+            builder.endArray();
+        }
+        builder.endObject();
+        return new RestResponse(RestStatus.OK, CONTENT_TYPE, Strings.toString(builder));
+    }
+
+}

x-pack/plugin/prometheus/src/test/java/org/elasticsearch/xpack/prometheus/rest/PrometheusLabelsResponseListenerTests.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.prometheus.rest;
+
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.rest.FakeRestChannel;
+import org.elasticsearch.test.rest.FakeRestRequest;
+
+import java.util.List;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+
+public class PrometheusLabelsResponseListenerTests extends ESTestCase {
+
+    public void testCollectLabelNamesStripsLabelsPrefix() {
+        List<String> names = PrometheusLabelsResponseListener.collectLabelNames(
+            List.of(List.of("labels.__name__"), List.of("labels.job"), List.of("labels.instance"))
+        );
+        assertThat(names, equalTo(List.of("__name__", "job", "instance")));
+    }
+
+    public void testCollectLabelNamesSingleEntry() {
+        List<String> names = PrometheusLabelsResponseListener.collectLabelNames(List.of(List.of("labels.__name__")));
+        assertThat(names, equalTo(List.of("__name__")));
+    }
+
+    public void testCollectLabelNamesSkipsNullValues() {
+        List<List<Object>> rows = new java.util.ArrayList<>();
+        rows.add(java.util.Arrays.asList((Object) null));
+        rows.add(List.of("labels.job"));
+        assertThat(PrometheusLabelsResponseListener.collectLabelNames(rows), equalTo(List.of("job")));
+    }
+
+    public void testCollectLabelNamesEmptyRowsReturnsEmptyList() {
+        assertThat(PrometheusLabelsResponseListener.collectLabelNames(List.of()).isEmpty(), is(true));
+    }
+
+    public void testCollectLabelNamesOtelFieldsReturnedAsIs() {
+        // OTel dimension fields carry no "labels." prefix
+        List<String> names = PrometheusLabelsResponseListener.collectLabelNames(
+            List.of(List.of("attributes.service.name"), List.of("attributes.host.name"))
+        );
+        assertThat(names, equalTo(List.of("attributes.service.name", "attributes.host.name")));
+    }
+
+    public void testCollectLabelNamesMixedPrefixAndOtel() {
+        // Prometheus and OTel dimension field names in the same result set
+        List<String> names = PrometheusLabelsResponseListener.collectLabelNames(
+            List.of(List.of("labels.__name__"), List.of("labels.job"), List.of("attributes.service.name"))
+        );
+        assertThat(names, equalTo(List.of("__name__", "job", "attributes.service.name")));
+    }
+
+    public void testBuildSuccessResponseAlwaysEmitsNameLabelFirst() throws Exception {
+        // OTel case: __name__ is absent from the ESQL result — injected by serialisation
+        String body = PrometheusLabelsResponseListener.buildSuccessResponse(List.of("job", "env"), 0).content().utf8ToString();
+        assertThat(body, containsString("\"__name__\""));
+        // __name__ must appear before other labels in the JSON array
+        assertThat(body.indexOf("\"__name__\"") < body.indexOf("\"env\""), is(true));
+        assertThat(body.indexOf("\"__name__\"") < body.indexOf("\"job\""), is(true));
+    }
+
+    public void testBuildSuccessResponseDeduplicatesNameLabelForPrometheus() throws Exception {
+        // Prometheus case: __name__ comes through from the ESQL plan — must not appear twice
+        String body = PrometheusLabelsResponseListener.buildSuccessResponse(List.of("__name__", "job"), 0).content().utf8ToString();
+        int first = body.indexOf("\"__name__\"");
+        int second = body.indexOf("\"__name__\"", first + 1);
+        assertThat("__name__ should appear exactly once", second, is(-1));
+    }
+
+    public void testBuildSuccessResponseBodyShape() throws Exception {
+        String body = PrometheusLabelsResponseListener.buildSuccessResponse(List.of("job"), 0).content().utf8ToString();
+        assertThat(body, containsString("\"status\":\"success\""));
+        assertThat(body, containsString("\"__name__\""));
+        assertThat(body, containsString("\"job\""));
+    }
+
+    public void testNoWarningWhenResultsBelowLimit() throws Exception {
+        String body = PrometheusLabelsResponseListener.buildSuccessResponse(List.of("env", "job"), 5).content().utf8ToString();
+        assertThat(body, not(containsString("warnings")));
+    }
+
+    public void testWarningWhenResultsEqualLimit() throws Exception {
+        // size == limit signals possible truncation (plan applied Limit node)
+        String body = PrometheusLabelsResponseListener.buildSuccessResponse(List.of("env", "job"), 2).content().utf8ToString();
+        assertThat(body, containsString("warnings"));
+        assertThat(body, containsString("results truncated due to limit"));
+    }
+
+    public void testNoWarningWhenLimitIsZero() throws Exception {
+        String body = PrometheusLabelsResponseListener.buildSuccessResponse(List.of("env", "job"), 0).content().utf8ToString();
+        assertThat(body, not(containsString("warnings")));
+    }
+
+    public void testOnFailureBadRequest() throws Exception {
+        FakeRestRequest fakeRequest = new FakeRestRequest();
+        FakeRestChannel channel = new FakeRestChannel(fakeRequest, true);
+        var listener = PrometheusLabelsResponseListener.create(channel, 0);
+
+        ElasticsearchStatusException ex = new ElasticsearchStatusException("bad selector syntax", RestStatus.BAD_REQUEST);
+        listener.onFailure(ex);
+
+        assertThat(channel.errors().get(), is(1));
+        assertThat(channel.capturedResponse().status(), equalTo(RestStatus.BAD_REQUEST));
+    }
+
+    public void testOnFailureInternalError() throws Exception {
+        FakeRestRequest fakeRequest = new FakeRestRequest();
+        FakeRestChannel channel = new FakeRestChannel(fakeRequest, true);
+        var listener = PrometheusLabelsResponseListener.create(channel, 0);
+
+        listener.onFailure(new RuntimeException("something went wrong"));
+
+        assertThat(channel.errors().get(), is(1));
+        assertThat(channel.capturedResponse().status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
+    }
+
+    public void testOnFailureResponseBodyContainsErrorType() throws Exception {
+        FakeRestRequest fakeRequest = new FakeRestRequest();
+        FakeRestChannel channel = new FakeRestChannel(fakeRequest, true);
+        var listener = PrometheusLabelsResponseListener.create(channel, 0);
+
+        ElasticsearchStatusException ex = new ElasticsearchStatusException("invalid parameter", RestStatus.BAD_REQUEST);
+        listener.onFailure(ex);
+
+        String body = channel.capturedResponse().content().utf8ToString();
+        assertThat(body, containsString("\"status\":\"error\""));
+        assertThat(body, containsString("\"errorType\":\"bad_data\""));
+        assertThat(body, containsString("invalid parameter"));
+    }
+}