Logging improvements: informing which ES version considered when building this plugin version, when plugin and connected ES versions do not match warn/inform incompatibility risk and show the guidance.

Author: mashhurs

lib/logstash/filters/elastic_integration.rb

@@ -135,6 +135,10 @@ def register
     initialize_event_processor!
 
     perform_preflight_check!
+    if serverless?
+      set_api_version_to_rest_client!
+    end
+    check_versions_alignment
   end # def register
 
   def filter(event)
@@ -342,12 +346,6 @@ def extract_immutable_config
   def initialize_elasticsearch_rest_client!
     config = extract_immutable_config
     @elasticsearch_rest_client = _elasticsearch_rest_client(config)
-
-    if serverless?
-      @elasticsearch_rest_client = _elasticsearch_rest_client(config) do |builder|
-                                    builder.configureElasticApi { |elasticApi| elasticApi.setApiVersion(ELASTIC_API_VERSION) }
-                                  end
-    end
   end
 
   def _elasticsearch_rest_client(config, &builder_interceptor)
@@ -374,13 +372,21 @@ def initialize_event_processor!
   end
 
   def perform_preflight_check!
+    java_import('co.elastic.logstash.filters.elasticintegration.PreflightCheck')
+    @preflight_check = PreflightCheck.new(@elasticsearch_rest_client)
+    connected_es_version_info
     check_user_privileges!
     check_es_cluster_license!
+  rescue => e
+    raise_config_error!(e.message)
+  end
+
+  def connected_es_version_info
+    @connected_es_version_info |= @preflight_check.getElasticsearchVersionInfo
   end
 
   def check_user_privileges!
-    java_import('co.elastic.logstash.filters.elasticintegration.PreflightCheck')
-    PreflightCheck.new(@elasticsearch_rest_client).checkUserPrivileges
+    @preflight_check.checkUserPrivileges
   rescue => e
     security_error_message = "no handler found for uri [/_security/user/_has_privileges]"
     if e.message.include?(security_error_message)
@@ -403,17 +409,19 @@ def check_user_privileges!
   end
 
   def check_es_cluster_license!
-    java_import('co.elastic.logstash.filters.elasticintegration.PreflightCheck')
-    PreflightCheck.new(@elasticsearch_rest_client).checkLicense
+    @preflight_check.checkLicense
   rescue => e
     raise_config_error!(e.message)
   end
 
   def serverless?
-    java_import('co.elastic.logstash.filters.elasticintegration.PreflightCheck')
-    PreflightCheck.new(@elasticsearch_rest_client).isServerless
-  rescue => e
-    raise_config_error!(e.message)
+    connected_es_version_info["build_flavor"] == 'serverless'
+  end
+
+  def set_api_version_to_rest_client!
+    @elasticsearch_rest_client = _elasticsearch_rest_client(config) do |builder|
+      builder.configureElasticApi { |elasticApi| elasticApi.setApiVersion(ELASTIC_API_VERSION) }
+    end
   end
 
   ##
@@ -454,4 +462,40 @@ def ensure_java_major_version!(minimum_major_version)
       ERR
     end
   end
+
+  ##
+  # compares the current plugin version with the Elasticsearch version connected to
+  # generates a warning or info message based on the situation where the plugin is ahead or behind of the Elasticsearch
+  def check_versions_alignment
+    plugin_version_parts = VERSION.split('.')
+    plugin_major_version = plugin_version_parts[0].to_i
+    plugin_minor_version = plugin_version_parts[1].to_i
+
+    base_message = "This #{VERSION} version of plugin embedded Ingest node components from Elasticsearch #{plugin_major_version}.#{plugin_minor_version}"
+    @logger.info(base_message)
+
+    es_version_parts = connected_es_version_info["number"].split('.')
+    es_major_version = es_version_parts[0].to_i
+    es_minor_version = es_version_parts[1].to_i
+
+    return if (plugin_major_version == es_major_version && plugin_minor_version == es_minor_version)
+
+    descriptive_message = "Upgrade the plugin and/or stack to the same `major.minor` to get the minimal disruptive experience"
+
+    if plugin_major_version != es_major_version
+      # major version difference may alarm the warning, it doesn't seem useful to inform minor here
+      # when plugin is ahead of connected ES, not much concern but recommended to keep the same version
+      # when plugin is behind of connected ES, BIG concern as it cannot utilize the features of the Ingest Node
+      period_indicator = plugin_major_version > es_major_version ? "newer" : "older"
+      message = "This #{VERSION} version of plugin is compiled with #{period_indicator} Elasticsearch version than" +
+        " currently connected Elasticsearch #{es_major_version}.#{es_minor_version} version. #{descriptive_message}"
+      @logger.info(message) if plugin_major_version > es_major_version
+      @logger.warn(message) if plugin_major_version < es_major_version
+    else
+      period_indicator = plugin_minor_version > es_minor_version ? "newer" : "older"
+      message = "This #{VERSION} version of plugin is compiled with #{period_indicator} Elasticsearch version than" +
+        " currently connected Elasticsearch #{es_major_version}.#{es_minor_version} version. #{descriptive_message}"
+      @logger.info(message) unless plugin_minor_version == es_minor_version
+    end
+  end
 end

spec/unit/elastic_integration_spec.rb

@@ -29,7 +29,7 @@
         allow(java.lang.System).to receive(:getProperty).with("java.specification.version").and_return("11.0.16.1")
       end
 
-      it 'prevents initialization and presents helpful guidancee' do
+      it 'prevents initialization and presents helpful guidance' do
         expect { described_class.new({}) }.to raise_error(LogStash::EnvironmentError)
                                                 .with_message(including("requires Java 21 or later", # reason +
                                                                         "current JVM version `11.0.16.1`",
@@ -65,7 +65,9 @@
 
   describe "plugin register" do
 
-    before(:each) { allow(plugin).to receive(:perform_preflight_check!).and_return(true) }
+    before(:each) { allow(plugin).to receive(:connected_es_version_info).and_return({'number' => '8.17.0', 'build_flavor' => 'default'}) }
+    before(:each) { allow(plugin).to receive(:check_user_privileges!).and_return(true) }
+    before(:each) { allow(plugin).to receive(:check_es_cluster_license!).and_return(true) }
     before(:each) { allow(plugin).to receive(:serverless?).and_return(false) }
 
     let(:registered_plugin) { plugin.tap(&:register) }

src/main/java/co/elastic/logstash/filters/elasticintegration/PreflightCheck.java

@@ -50,6 +50,38 @@ public PreflightCheck(final RestClient elasticsearchRestClient) {
         this.elasticsearchRestClient = elasticsearchRestClient;
     }
 
+    /**
+     * Get Elasticsearch cluster info and store version number and build flavor.
+     * @return {@link Map} containing
+     * <ul>
+     *   <li>"number": the key for a version number (e.g., "8.17.0")</li>
+     *   <li>"build_flavor": the key for a build flavor (e.g., "default" "oss" or "serverless")</li>
+     * </ul>
+     * Throws {@link Failure} if an error occurs while sending the request, parsing the response or extracting the
+     *  required information.
+     */
+    public Map<String, String> getElasticsearchVersionInfo() {
+        try {
+            final Request request = new Request("GET", "/");
+            final Response response = elasticsearchRestClient.performRequest(request);
+
+            final String resBody = new String(response.getEntity().getContent().readAllBytes(), StandardCharsets.UTF_8);
+            logger.debug(() -> String.format("Elasticsearch '/' RAW: %s", resBody));
+
+            final JsonNode versionNode = OBJECT_MAPPER.readTree(resBody).get("version");
+
+            final String elasticsearchVersion = versionNode.get("number").asText();
+            logger.info(String.format("Connected to Elasticsearch version: %s", elasticsearchVersion));
+
+            final String elasticsearchBuildFlavor = versionNode.get("build_flavor").asText();
+            logger.info(String.format("Elasticsearch build_flavor: %s", elasticsearchBuildFlavor));
+
+            return Map.of("number", elasticsearchVersion, "build_flavor", elasticsearchBuildFlavor);
+        } catch (Exception e) {
+            throw new Failure(String.format("Fetching Elasticsearch version information failed: %s", e.getMessage()), e);
+        }
+    }
+
     public void checkUserPrivileges() {
         try {
             final Request hasPrivilegesRequest = new Request("POST", "/_security/user/_has_privileges");
@@ -105,26 +137,6 @@ public void checkLicense() {
         }
     }
 
-    public boolean isServerless() {
-        try {
-            final Request req = new Request("GET", "/");
-            final Response res = elasticsearchRestClient.performRequest(req);
-
-            final String resBody = new String(res.getEntity().getContent().readAllBytes(), StandardCharsets.UTF_8);
-            logger.debug(() -> String.format("Elasticsearch '/' RAW: %s", resBody));
-
-            final JsonNode versionNode = OBJECT_MAPPER.readTree(resBody).get("version");
-
-            final String buildFlavor = versionNode.get("build_flavor").asText();
-            logger.info(String.format("Elasticsearch build_flavor: %s", buildFlavor));
-
-            return buildFlavor.equals("serverless");
-        } catch (Exception e) {
-            logger.error(String.format("Exception checking serverless: %s", e.getMessage()));
-            throw new Failure(String.format("Preflight check failed: %s", e.getMessage()), e);
-        }
-    }
-    
     public static class Failure extends RuntimeException {
         public Failure(String message, Throwable cause) {
             super(message, cause);

src/test/java/co/elastic/logstash/filters/elasticintegration/ElasticsearchRestClientBuilderTest.java

@@ -156,6 +156,8 @@ public void testElasticApiConfigAddsHeaders() throws HttpException, IOException
             for (HttpRequestInterceptor interceptor : interceptors) {
                 interceptor.process(request, null);
             }
+            assertThat(request.getFirstHeader("Elastic-Api-Version").getValue(),
+                    is("2023-10-31"));
             assertThat(request.getFirstHeader("x-elastic-product-origin").getValue(),
                     is("logstash-filter-elastic_integration"));
         }

src/test/java/co/elastic/logstash/filters/elasticintegration/PreflightCheckTest.java

@@ -6,7 +6,6 @@
  */
 package co.elastic.logstash.filters.elasticintegration;
 
-import co.elastic.logstash.api.Password;
 import com.github.tomakehurst.wiremock.extension.responsetemplating.ResponseTemplateTransformer;
 import com.github.tomakehurst.wiremock.http.Fault;
 import com.github.tomakehurst.wiremock.junit5.WireMockExtension;
@@ -19,6 +18,7 @@
 import java.net.URL;
 import java.nio.file.Path;
 import java.util.Collections;
+import java.util.Map;
 import java.util.function.Consumer;
 
 import static co.elastic.logstash.filters.elasticintegration.util.ResourcesUtil.readResource;
@@ -77,7 +77,6 @@ void checkCredentialsPrivilegesMissingReadPipeline() throws Exception {
     }
 
 
-
     @Test
     void checkCredentialsPrivilegesMissingManageIndexTemplates() throws Exception {
         wireMock.stubFor(
@@ -197,57 +196,50 @@ void checkLicenseConnectionError() throws Exception {
     }
 
     @Test
-    void checkIsServerlessTrue() throws Exception {
+    void checkServerlessVersionInfo() throws Exception {
         wireMock.stubFor(get("/")
-                .willReturn(okJson(getBodyFixture("is_serverless.true.json"))
+                .willReturn(okJson(getBodyFixture("serverless.response.json"))
                         .withTransformers("response-template")));
         withWiremockElasticsearch((restClient -> {
             final Logger logger = Mockito.mock(Logger.class);
-            boolean result = new PreflightCheck(logger, restClient).isServerless();
+            Map<String, String> result = new PreflightCheck(logger, restClient).getElasticsearchVersionInfo();
 
-            assertTrue(result);
+            assertNotNull(result);
+            assertThat(result.size(), equalTo(2));
+            assertEquals(result.get("number"), "8.11.0");
+            assertEquals(result.get("build_flavor"), "serverless");
+            Mockito.verify(logger).info(argThat(containsString("Connected to Elasticsearch version: 8.11.0")));
             Mockito.verify(logger).info(argThat(containsString("Elasticsearch build_flavor: serverless")));
         }));
     }
 
     @Test
-    void checkIsServerlessFalse() throws Exception {
+    void testElasticsearchVersionInfo() throws Exception {
         wireMock.stubFor(get("/")
-                .willReturn(okJson(getBodyFixture("is_serverless.false.json"))
+                .willReturn(okJson(getBodyFixture("elasticsearch.default.response.json"))
                         .withTransformers("response-template")));
         withWiremockElasticsearch((restClient -> {
             final Logger logger = Mockito.mock(Logger.class);
-            boolean result = new PreflightCheck(logger, restClient).isServerless();
+            Map<String, String> versionInfo = new PreflightCheck(logger, restClient).getElasticsearchVersionInfo();
 
-            assertFalse(result);
+            assertNotNull(versionInfo);
+            assertThat(versionInfo.size(), equalTo(2));
+            assertEquals(versionInfo.get("number"), "8.7.0");
+            assertEquals(versionInfo.get("build_flavor"), "default");
+            Mockito.verify(logger).info(argThat(containsString("Connected to Elasticsearch version: 8.7.0")));
             Mockito.verify(logger).info(argThat(containsString("Elasticsearch build_flavor: default")));
         }));
     }
 
     @Test
-    void checkIsServerless401Response() throws Exception {
+    void testElasticsearch401Response() throws Exception {
         wireMock.stubFor(get("/")
-                        .willReturn(jsonResponse(getBodyFixture("is_serverless.resp.401.json"), 401)));
+                        .willReturn(jsonResponse(getBodyFixture("elasticsearch.401.response.json"), 401)));
         withWiremockElasticsearch((restClient -> {
             final PreflightCheck.Failure failure = assertThrows(PreflightCheck.Failure.class, () -> {
-                new PreflightCheck(restClient).isServerless();
+                new PreflightCheck(restClient).getElasticsearchVersionInfo();
             });
-            assertThat(failure.getMessage(), hasToString(stringContainsInOrder("Preflight check failed", "401 Unauthorized")));
-        }));
-    }
-
-    public static final String ELASTIC_API_VERSION = "2023-10-31";
-    public static final String ENCODED_API_KEY = "iamapikeyiamapikeyiamapikeyiamapikeyiamapikeyiamapikeyiama==";
-    @Test
-    void checkServerlessRequestContainsHeaders() throws Exception {
-        wireMock.stubFor(get("/")
-                        .withHeader("Elastic-Api-Version", containing(ELASTIC_API_VERSION))
-                        .withHeader("Authorization", containing(ENCODED_API_KEY))
-                .willReturn(okJson(getBodyFixture("is_serverless.true.json"))
-                        .withTransformers("response-template")));
-        withWiremockServerlessElasticsearch((restClient -> {
-            boolean result = new PreflightCheck(restClient).isServerless();
-            assertTrue(result);
+            assertThat(failure.getMessage(), hasToString(stringContainsInOrder("Fetching Elasticsearch version information failed", "401 Unauthorized")));
         }));
     }
 
@@ -258,17 +250,6 @@ private void withWiremockElasticsearch(final Consumer<RestClient> handler) throw
         }
     }
 
-    private void withWiremockServerlessElasticsearch(final Consumer<RestClient> handler) throws Exception{
-        final URL wiremockElasticsearch = new URL("http", "127.0.0.1", wireMock.getRuntimeInfo().getHttpPort(),"/");
-        try (RestClient restClient = ElasticsearchRestClientBuilder
-                .forURLs(Collections.singletonList(wiremockElasticsearch))
-                .configureElasticApi(c -> c.setApiVersion(ELASTIC_API_VERSION))
-                .configureRequestAuth(c -> c.setApiKey(new Password(ENCODED_API_KEY)))
-                .build()) {
-            handler.accept(restClient);
-        }
-    }
-
     static String getBodyFixture(final String name) {
         return readResource(PreflightCheck.class, Path.of("preflight-check", name).toString());
     }