Get the Logstash version via API

src/main/java/co/elastic/support/diagnostics/chain/DiagnosticChainExec.java

@@ -9,6 +9,7 @@
 import co.elastic.support.diagnostics.DiagnosticException;
 import co.elastic.support.diagnostics.commands.CheckDiagnosticVersion;
 import co.elastic.support.diagnostics.commands.CheckElasticsearchVersion;
+import co.elastic.support.diagnostics.commands.CheckLogstashVersion;
 import co.elastic.support.diagnostics.commands.CheckKibanaVersion;
 import co.elastic.support.diagnostics.commands.CheckPlatformDetails;
 import co.elastic.support.diagnostics.commands.CheckUserAuthLevel;
@@ -70,6 +71,7 @@ public static void runDiagnostic(DiagnosticContext context, String type) throws
                 break;
 
             case Constants.logstashLocal:
+                new CheckLogstashVersion().execute(context);
                 new RunLogstashQueries().execute(context);
                 new GenerateLogstashDiagnostics().execute(context);
                 if (context.runSystemCalls) {
@@ -82,6 +84,7 @@ public static void runDiagnostic(DiagnosticContext context, String type) throws
                 break;
 
             case Constants.logstashRemote:
+                new CheckLogstashVersion().execute(context);
                 new RunLogstashQueries().execute(context);
                 new GenerateLogstashDiagnostics().execute(context);
                 if (context.runSystemCalls) {
@@ -93,6 +96,7 @@ public static void runDiagnostic(DiagnosticContext context, String type) throws
                 break;
 
             case Constants.logstashApi:
+                new CheckLogstashVersion().execute(context);
                 new RunLogstashQueries().execute(context);
                 new GenerateLogstashDiagnostics().execute(context);
                 break;

src/main/java/co/elastic/support/diagnostics/commands/CheckLogstashVersion.java

@@ -0,0 +1,119 @@
+/*
+ * 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 co.elastic.support.diagnostics.commands;
+
+import co.elastic.support.Constants;
+import co.elastic.support.diagnostics.DiagnosticException;
+import co.elastic.support.diagnostics.chain.Command;
+import co.elastic.support.diagnostics.chain.DiagnosticContext;
+import co.elastic.support.rest.RestClient;
+import co.elastic.support.rest.RestEntryConfig;
+import co.elastic.support.rest.RestResult;
+import co.elastic.support.util.JsonYamlUtils;
+import co.elastic.support.util.SystemProperties;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.semver4j.Semver;
+import org.semver4j.SemverException;
+
+import java.util.Map;
+
+/**
+ * {@code CheckLogstashVersion} uses the REST configuration to fetch the version
+ * of
+ * Logstash from the server.
+ *
+ * If this request fails, then the rest of the diagnostic cannot process because
+ * REST
+ * calls are setup against specific versions and, without having a version, they
+ * cannot
+ * be setup.
+ */
+public class CheckLogstashVersion implements Command {
+
+    private static final Logger logger = LogManager.getLogger(CheckLogstashVersion.class);
+
+    public void execute(DiagnosticContext context) throws DiagnosticException {
+
+        // Get the version number from the JSON returned
+        // by just submitting the host/port combo
+        logger.info(Constants.CONSOLE, "Getting Logstash Version.");
+
+        try {
+            RestClient restClient = RestClient.getClient(
+                    context.diagnosticInputs.host,
+                    context.diagnosticInputs.port,
+                    context.diagnosticInputs.scheme,
+                    context.diagnosticInputs.user,
+                    context.diagnosticInputs.password,
+                    context.diagnosticInputs.proxyHost,
+                    context.diagnosticInputs.proxyPort,
+                    context.diagnosticInputs.proxyUser,
+                    context.diagnosticInputs.proxyPassword,
+                    context.diagnosticInputs.pkiKeystore,
+                    context.diagnosticInputs.pkiKeystorePass,
+                    context.diagnosticInputs.skipVerification,
+                    context.diagsConfig.extraHeaders,
+                    context.diagsConfig.connectionTimeout,
+                    context.diagsConfig.connectionRequestTimeout,
+                    context.diagsConfig.socketTimeout);
+
+            // Add it to the global cache - automatically closed on exit.
+            context.resourceCache.addRestClient(Constants.restInputHost, restClient);
+            context.version = getLogstashVersion(restClient);
+            String version = context.version.getVersion();
+            RestEntryConfig builder = new RestEntryConfig(version);
+            Map restCalls = JsonYamlUtils.readYamlFromClasspath(Constants.LS_REST, true);
+
+            // logger.info(Constants.CONSOLE, restCalls);
+            logger.info(Constants.CONSOLE, "Run basic queries for Logstash: {}", restCalls);
+
+            context.elasticRestCalls = builder.buildEntryMap(restCalls);
+            context.fullElasticRestCalls = context.elasticRestCalls;
+        } catch (Exception e) {
+            logger.error("Unanticipated error:", e);
+            String errorLog = "Couldn't retrieve Logstash version due to a system or network error. %s%s%s";
+            errorLog = String.format(errorLog, e.getMessage(),
+                    SystemProperties.lineSeparator,
+                    Constants.CHECK_LOG);
+            throw new DiagnosticException(errorLog);
+        }
+    }
+
+    /**
+     * Fetch the Logstash version using the {@code client}, which is used to then to
+     * determine which
+     * REST endpoints can be used from the diagnostic.
+     *
+     * @param client The configured client to connect to Logstash.
+     * @return The Logstash version (semver).
+     * @throws DiagnosticException if the request fails or the version is invalid
+     */
+    public static Semver getLogstashVersion(RestClient client) throws DiagnosticException {
+        RestResult res = client.execQuery("/");
+        if (!res.isValid()) {
+            throw new DiagnosticException(
+                    res.formatStatusMessage("Could not retrieve the Logstash version - unable to continue."));
+        }
+        String result = res.toString();
+        JsonNode root = JsonYamlUtils.createJsonNodeFromString(result);
+        String version = root.path("version").asText();
+
+        logger.info(Constants.CONSOLE, String.format("Logstash Version is :%s", version));
+
+        try {
+            // Workaround for the semver issue with pre-release versions
+            // https://github.com/semver4j/semver4j/issues/307
+            return new Semver(version).withClearedPreRelease();
+        } catch (SemverException ex) {
+            throw new DiagnosticException(
+                    String.format("Logstash version format is wrong - unable to continue. (%s)", version));
+        }
+    }
+
+}

src/main/java/co/elastic/support/diagnostics/commands/RunLogstashQueries.java

@@ -41,14 +41,8 @@ public void execute(DiagnosticContext context) throws DiagnosticException {
 
         try {
             RestClient client = context.resourceCache.getRestClient(Constants.restInputHost);
-
-            RestEntryConfig builder = new RestEntryConfig("1.0.0");
-            Map restCalls = JsonYamlUtils.readYamlFromClasspath(Constants.LS_REST, true);
-            Map<String, RestEntry> entries = builder.buildEntryMap(restCalls);
-
-            List<RestEntry> queries = new ArrayList<>();
-            queries.addAll(entries.values());
-            runQueries(client, queries, context.tempDir, 0, 0);
+            List<RestEntry> entries = new ArrayList<>(context.elasticRestCalls.values());
+            runQueries(client, entries, context.tempDir, 0, 0);
 
             // Get the information we need to run system calls. It's easier to just get it
             // off disk after all the REST calls run.

src/main/resources/logstash-rest.yml

@@ -20,25 +20,25 @@ logstash_health_report:
 
 logstash_node:
   versions:
-    "> 0.0.0": "/_node"
+    "> 5.0.0": "/_node"
 
 logstash_nodes_hot_threads:
   versions:
-    "> 0.0.0": "/_node/hot_threads?threads=10000"
+    "> 5.0.0": "/_node/hot_threads?threads=10000"
 
 logstash_nodes_hot_threads_human:
   extension: .txt
   versions:
-    "> 0.0.0": "/_node/hot_threads?human&threads=10000"
+    "> 5.0.0": "/_node/hot_threads?human&threads=10000"
 
 logstash_node_stats:
   versions:
-    "> 0.0.0": "/_node/stats"
+    "> 5.0.0": "/_node/stats"
 
 logstash_plugins:
   versions:
-    "> 0.0.0": "/_node/plugins"
+    "> 5.0.0": "/_node/plugins"
 
 logstash_version:
   versions:
-    "> 0.0.0": "/"
+    "> 5.0.0": "/"

src/test/java/co/elastic/support/diagnostics/commands/TestCheckLogstashVersionTest.java

@@ -0,0 +1,139 @@
+package co.elastic.support.diagnostics.commands;
+
+import co.elastic.support.diagnostics.DiagnosticException;
+import org.junit.jupiter.api.*;
+import org.mockserver.integration.ClientAndServer;
+import co.elastic.support.rest.RestClient;
+import org.semver4j.Semver;
+import java.util.Collections;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockserver.integration.ClientAndServer.startClientAndServer;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class TestCheckLogstashVersionTest {
+    private ClientAndServer mockServer;
+    private RestClient httpRestClient;
+
+    @BeforeAll
+    public void globalSetup() {
+        mockServer = startClientAndServer(9881);
+    }
+
+    @AfterAll
+    public void globalTeardown() {
+        mockServer.stop();
+    }
+
+    @BeforeEach
+    public void setup() {
+        httpRestClient = RestClient.getClient(
+            "localhost",
+            9881,
+            "http",
+            "elastic",
+            "elastic",
+            "",
+            0,
+            "",
+            "",
+            "",
+            "",
+            true,
+            Collections.emptyMap(),
+            3000,
+            3000,
+            3000
+        );
+    }
+
+    @AfterEach
+    public void tearDown() {
+        mockServer.reset();
+    }
+
+    private void initializeLogstashMainHandler(String version) {
+        mockServer
+            .when(
+                request()
+                    .withMethod("GET")
+                    .withPath("/")
+            )
+            .respond(
+                response()
+                    .withBody("{\"host\":\"Test\",\"version\":\"" + version + "\",\"http_address\":\"127.0.0.1:9600\",\"id\":\"9ac54ae7-377e-4352-9727-15db6344332a\",\"name\":\"LucaMBP\",\"ephemeral_id\":\"3f1d87db-07c0-4015-941a-4005bbf908fc\",\"snapshot\":false,\"status\":\"yellow\",\"pipeline\":{\"workers\":11,\"batch_size\":125,\"batch_delay\":50},\"build_date\":\"2025-06-17T14:07:37+00:00\",\"build_sha\":\"01b7a2d93e4cf143d4964c71259655cf4575b709\",\"build_snapshot\":false}")
+                    .withStatusCode(200)
+            );
+    }
+
+    @Test
+    public void testQueriesForLogstashVersionNormal() throws DiagnosticException {
+        initializeLogstashMainHandler("8.1.2");
+        Semver version = new CheckLogstashVersion().getLogstashVersion(httpRestClient);
+        assertEquals("8.1.2", version.getVersion());
+    }
+
+    @Test
+    public void testQueriesForLogstashVersionWithRC() throws DiagnosticException {
+        initializeLogstashMainHandler("9.0.0-beta1");
+        Semver version = new CheckLogstashVersion().getLogstashVersion(httpRestClient);
+        assertEquals("9.0.0", version.getVersion());
+    }
+
+    @Test
+    public void testQueriesForLogstashEmptyVersion() {
+        mockServer
+            .when(
+                request()
+                    .withMethod("GET")
+                    .withPath("/")
+            )
+            .respond(
+                response()
+                    .withBody("{}")
+                    .withStatusCode(200)
+            );
+
+        try {
+            Semver version = new CheckLogstashVersion().getLogstashVersion(httpRestClient);
+            assertTrue(false);
+        } catch (DiagnosticException e) {
+            assertEquals("Logstash version format is wrong - unable to continue. ()", e.getMessage());
+        }
+    }
+
+    @Test
+    public void testQueriesForLogstashCorruptedVersion() {
+        initializeLogstashMainHandler("a.v.c");
+        try {
+            Semver version = new CheckLogstashVersion().getLogstashVersion(httpRestClient);
+            assertTrue(false);
+        } catch (DiagnosticException e) {
+            assertEquals("Logstash version format is wrong - unable to continue. (a.v.c)", e.getMessage());
+        }
+    }
+
+    @Test
+    public void testQueriesForLogstashTextWithVersion() {
+        initializeLogstashMainHandler("test-6.5.1");
+        mockServer
+            .when(
+                request()
+                    .withMethod("GET")
+                    .withPath("/")
+            )
+            .respond(
+                response()
+                    .withBody("{\"version\":\"test-6.5.1\"}")
+                    .withStatusCode(200)
+            );
+
+        try {
+            Semver version = new CheckLogstashVersion().getLogstashVersion(httpRestClient);
+            assertTrue(false);
+        } catch (DiagnosticException e) {
+            assertEquals("Logstash version format is wrong - unable to continue. (test-6.5.1)", e.getMessage());
+        }
+    }
+}