Fix an authorization error when LogsPatternUsageService attempts to update `logsdb.prior_logs_usage` cluster setting (#128050)

The following failure occurs when the LogsPatternUsageService updates the `logsb.prior_logs_usage` cluster setting:

```
[2025-05-13T17:11:21,685][DEBUG ][o.e.x.l.LogsPatternUsageService] [test-cluster-0] Failed to update [logsdb.prior_logs_usage] org.elasticsearch.ElasticsearchSecurityException: action [cluster:admin/settings/update] is unauthorized for user [_system] with effective roles [_system], this action is granted by the cluster privileges [manage,all]
	at org.elasticsearch.xcore@8.19.0-SNAPSHOT/org.elasticsearch.xpack.core.security.support.Exceptions.authorizationError(Exceptions.java:36)
	at org.elasticsearch.security@8.19.0-SNAPSHOT/org.elasticsearch.xpack.security.authz.AuthorizationService.denialException(AuthorizationService.java:1014)
	at org.elasticsearch.security@8.19.0-SNAPSHOT/org.elasticsearch.xpack.security.authz.AuthorizationService.actionDenied(AuthorizationService.java:991)
	at org.elasticsearch.security@8.19.0-SNAPSHOT/org.elasticsearch.xpack.security.authz.AuthorizationService.actionDenied(AuthorizationService.java:980)
	at org.elasticsearch.security@8.19.0-SNAPSHOT/org.elasticsearch.xpack.security.authz.AuthorizationService.actionDenied(AuthorizationService.java:970)
	at org.elasticsearch.security@8.19.0-SNAPSHOT/org.elasticsearch.xpack.security.authz.AuthorizationService.authorizeSystemUser(AuthorizationService.java:706)
	at org.elasticsearch.security@8.19.0-SNAPSHOT/org.elasticsearch.xpack.security.authz.AuthorizationService.authorize(AuthorizationService.java:320)
	at org.elasticsearch.security@8.19.0-SNAPSHOT/org.elasticsearch.xpack.security.action.filter.SecurityActionFilter.lambda$applyInternal$5(SecurityActionFilter.java:178)
```

This results in the `logsdb.prior_logs_usage` cluster setting not being set and causes the LogsPatternUsageService to continuously attempt to update the mentioned setting every 1 minute. The result of this is that the mentioned setting is never set, which if a cluster upgrades to 9.x, causes logsdb to be enabled by default. Which shouldn't happen for clusters upgrading from 8.x with data streams in the `logs-*-*` namespace.

Unfortunately, this bug wasn't noticed given that the error wasn't visible (only if DEBUG logging was enabled) and the tests that test this functionality specifically don't run with security enabled.

The fix is to use OriginSettingClient client instead of node client directly, so that xpack internal user is used, which does have to privileges to update cluster settings.

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 @@ public final class ClientHelper {
     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.ActionListener;
 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 org.elasticsearch.threadpool.ThreadPool;
 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 @@ final class LogsPatternUsageService implements LocalNodeMasterListener {
                 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.ClusterState;
 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 class LogsPatternUsageServiceTests extends ESTestCase {
         }).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 class LogsPatternUsageServiceTests extends ESTestCase {
         }).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 class LogsPatternUsageServiceTests extends ESTestCase {
         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 class LogsPatternUsageServiceTests extends ESTestCase {
         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 class LogsPatternUsageServiceTests extends ESTestCase {
         assertTrue(service.hasPriorLogsUsage);
         assertNull(service.cancellable);
 
-        verifyNoInteractions(client, threadPool);
+        verify(client, times(1)).threadPool();
+        verifyNoInteractions(threadPool);
     }
 
     public void testCheckHasUsageUnexpectedResponse() {
@@ -170,6 +176,8 @@ public class LogsPatternUsageServiceTests extends ESTestCase {
         }).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.IDP_ORIGIN;
 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 final class AuthorizationUtils {
             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;