Fix an authorization error when LogsPatternUsageService attempts to update logsdb.prior_logs_usage cluster setting

docs/changelog/128050.yaml

@@ -0,0 +1,5 @@
+pr: 128050
+summary: Fix an authorization error when LogsPatternUsageService attempts to update `logsdb.prior_logs_usage` cluster setting.
+area: Logs
+type: bug
+issues: []

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ClientHelper.java

@@ -196,6 +196,7 @@ private static String maybeRewriteSingleAuthenticationHeaderForVersion(
     public static final String APM_ORIGIN = "apm";
     public static final String OTEL_ORIGIN = "otel";
     public static final String REINDEX_DATA_STREAM_ORIGIN = "reindex_data_stream";
+    public static final String LOGS_PATTERN_USAGE_ORIGIN = "logs_pattern_usage";
 
     private ClientHelper() {}
 

x-pack/plugin/logsdb/src/javaRestTest/java/org/elasticsearch/xpack/logsdb/LogsdbWithSecurityRestIT.java

@@ -0,0 +1,84 @@
+/*
+ * 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.logsdb;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.common.settings.SecureString;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.elasticsearch.test.rest.ESRestTestCase;
+import org.junit.ClassRule;
+
+import java.util.Map;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
+
+public class LogsdbWithSecurityRestIT extends ESRestTestCase {
+
+    private static final String PASSWORD = "secret-test-password";
+
+    @ClassRule
+    public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
+        .distribution(DistributionType.DEFAULT)
+        .setting("logsdb.usage_check.max_period", "1s")
+        .setting("xpack.license.self_generated.type", "trial")
+        .setting("xpack.security.enabled", "true")
+        .setting("xpack.security.transport.ssl.enabled", "false")
+        .setting("xpack.security.http.ssl.enabled", "false")
+        .user("test_admin", PASSWORD, "superuser", true)
+        .build();
+
+    @Override
+    protected String getTestRestCluster() {
+        return cluster.getHttpAddresses();
+    }
+
+    @Override
+    protected Settings restClientSettings() {
+        String token = basicAuthHeaderValue("test_admin", new SecureString(PASSWORD.toCharArray()));
+        return Settings.builder().put(ThreadContext.PREFIX + ".Authorization", token).build();
+    }
+
+    public void testPriorLogsUsage() throws Exception {
+        {
+            var getClusterSettingsRequest = new Request("GET", "/_cluster/settings");
+            getClusterSettingsRequest.addParameter("flat_settings", "true");
+            var getClusterSettingResponse = (Map<?, ?>) entityAsMap(client().performRequest(getClusterSettingsRequest));
+            var persistentSettings = (Map<?, ?>) getClusterSettingResponse.get("persistent");
+            assertThat(persistentSettings.get("logsdb.prior_logs_usage"), nullValue());
+        }
+
+        var request = new Request("POST", "/logs-test-foo/_doc");
+        request.setJsonEntity("""
+            {
+                "@timestamp": "2020-01-01T00:00:00.000Z",
+                "host.name": "foo",
+                "message": "bar"
+            }
+            """);
+        assertOK(client().performRequest(request));
+
+        String index = DataStream.getDefaultBackingIndexName("logs-test-foo", 1);
+        var settings = (Map<?, ?>) ((Map<?, ?>) getIndexSettings(index).get(index)).get("settings");
+        assertNull(settings.get("index.mode"));
+        assertNull(settings.get("index.mapping.source.mode"));
+
+        assertBusy(() -> {
+            var getClusterSettingsRequest = new Request("GET", "/_cluster/settings");
+            getClusterSettingsRequest.addParameter("flat_settings", "true");
+            var getClusterSettingResponse = (Map<?, ?>) entityAsMap(client().performRequest(getClusterSettingsRequest));
+            var persistentSettings = (Map<?, ?>) getClusterSettingResponse.get("persistent");
+            assertThat(persistentSettings.get("logsdb.prior_logs_usage"), equalTo("true"));
+        });
+    }
+
+}

x-pack/plugin/logsdb/src/main/java/org/elasticsearch/xpack/logsdb/LogsPatternUsageService.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsAction;
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
 import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.cluster.LocalNodeMasterListener;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.regex.Regex;
@@ -26,6 +27,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
+import static org.elasticsearch.xpack.core.ClientHelper.LOGS_PATTERN_USAGE_ORIGIN;
 import static org.elasticsearch.xpack.logsdb.LogsdbIndexModeSettingsProvider.LOGS_PATTERN;
 
 /**
@@ -70,7 +72,7 @@ final class LogsPatternUsageService implements LocalNodeMasterListener {
     volatile Scheduler.Cancellable cancellable;
 
     LogsPatternUsageService(Client client, Settings nodeSettings, ThreadPool threadPool, Supplier<Metadata> metadataSupplier) {
-        this.client = client;
+        this.client = new OriginSettingClient(client, LOGS_PATTERN_USAGE_ORIGIN);
         this.nodeSettings = nodeSettings;
         this.threadPool = threadPool;
         this.metadataSupplier = metadataSupplier;
@@ -155,11 +157,11 @@ void updateSetting() {
                 hasPriorLogsUsage = true;
                 cancellable = null;
             } else {
-                LOGGER.debug(() -> "unexpected response [" + LOGSDB_PRIOR_LOGS_USAGE.getKey() + "]");
+                LOGGER.debug(() -> "unexpected response [" + LOGSDB_PRIOR_LOGS_USAGE.getKey() + "], retrying...");
                 scheduleNext(TimeValue.ONE_MINUTE);
             }
         }, e -> {
-            LOGGER.debug(() -> "Failed to update [" + LOGSDB_PRIOR_LOGS_USAGE.getKey() + "]", e);
+            LOGGER.warn(() -> "Failed to update [" + LOGSDB_PRIOR_LOGS_USAGE.getKey() + "], retrying...", e);
             scheduleNext(TimeValue.ONE_MINUTE);
         }));
     }

x-pack/plugin/logsdb/src/test/java/org/elasticsearch/xpack/logsdb/LogsPatternUsageServiceTests.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.test.ESTestCase;
@@ -49,6 +50,7 @@ public void testOnMaster() throws Exception {
         }).when(client).execute(same(ClusterUpdateSettingsAction.INSTANCE), any(), any());
 
         try (var threadPool = new TestThreadPool(getTestName())) {
+            when(client.threadPool()).thenReturn(threadPool);
             var clusterState = DataStreamTestHelper.getClusterStateWithDataStreams(List.of(new Tuple<>("logs-app1-prod", 1)), List.of());
             Supplier<Metadata> metadataSupplier = clusterState::metadata;
 
@@ -80,6 +82,8 @@ public void testCheckHasUsage() {
         }).when(client).execute(same(ClusterUpdateSettingsAction.INSTANCE), any(), any());
 
         var threadPool = mock(ThreadPool.class);
+        when(client.threadPool()).thenReturn(threadPool);
+        when(threadPool.getThreadContext()).thenReturn(new ThreadContext(Settings.EMPTY));
         var scheduledCancellable = mock(Scheduler.ScheduledCancellable.class);
         when(threadPool.schedule(any(), any(), any())).thenReturn(scheduledCancellable);
         var clusterState = DataStreamTestHelper.getClusterStateWithDataStreams(List.of(new Tuple<>("logs-app1-prod", 1)), List.of());
@@ -104,6 +108,7 @@ public void testCheckHasUsageNoMatch() {
         var client = mock(Client.class);
 
         var threadPool = mock(ThreadPool.class);
+        when(client.threadPool()).thenReturn(threadPool);
         var scheduledCancellable = mock(Scheduler.ScheduledCancellable.class);
         when(threadPool.schedule(any(), any(), any())).thenReturn(scheduledCancellable);
         var clusterState = DataStreamTestHelper.getClusterStateWithDataStreams(List.of(new Tuple<>("log-app1-prod", 1)), List.of());
@@ -120,7 +125,7 @@ public void testCheckHasUsageNoMatch() {
         assertEquals(service.nextWaitTime, TimeValue.timeValueMinutes(2));
 
         verify(threadPool, times(2)).schedule(any(), any(), any());
-        verifyNoInteractions(client);
+        verify(client, times(1)).threadPool();
     }
 
     public void testCheckPriorLogsUsageAlreadySet() {
@@ -148,7 +153,8 @@ public void testCheckPriorLogsUsageAlreadySet() {
         assertTrue(service.hasPriorLogsUsage);
         assertNull(service.cancellable);
 
-        verifyNoInteractions(client, threadPool);
+        verify(client, times(1)).threadPool();
+        verifyNoInteractions(threadPool);
     }
 
     public void testCheckHasUsageUnexpectedResponse() {
@@ -170,6 +176,8 @@ public void testCheckHasUsageUnexpectedResponse() {
         }).when(client).execute(same(ClusterUpdateSettingsAction.INSTANCE), any(), any());
 
         var threadPool = mock(ThreadPool.class);
+        when(threadPool.getThreadContext()).thenReturn(new ThreadContext(Settings.EMPTY));
+        when(client.threadPool()).thenReturn(threadPool);
         var scheduledCancellable = mock(Scheduler.ScheduledCancellable.class);
         when(threadPool.schedule(any(), any(), any())).thenReturn(scheduledCancellable);
         var clusterState = DataStreamTestHelper.getClusterStateWithDataStreams(List.of(new Tuple<>("logs-app1-prod", 1)), List.of());

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationUtils.java

@@ -37,6 +37,7 @@
 import static org.elasticsearch.xpack.core.ClientHelper.INDEX_LIFECYCLE_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.INFERENCE_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.LOGSTASH_MANAGEMENT_ORIGIN;
+import static org.elasticsearch.xpack.core.ClientHelper.LOGS_PATTERN_USAGE_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.MONITORING_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.OTEL_ORIGIN;
@@ -164,6 +165,7 @@ public static void switchUserBasedOnActionOriginAndExecute(
             case ENT_SEARCH_ORIGIN:
             case CONNECTORS_ORIGIN:
             case INFERENCE_ORIGIN:
+            case LOGS_PATTERN_USAGE_ORIGIN:
             case TASKS_ORIGIN:   // TODO use a more limited user for tasks
                 securityContext.executeAsInternalUser(InternalUsers.XPACK_USER, version, consumer);
                 break;