[Code] Use callWithInternalUser to manipulate worker queue index (#36715

)

x-pack/plugins/code/server/init.ts

@@ -30,6 +30,7 @@ import { IndexScheduler, UpdateScheduler } from './scheduler';
 import { CodeServerRouter } from './security';
 import { ServerOptions } from './server_options';
 import { ServerLoggerFactory } from './utils/server_logger_factory';
+import { EsClientWithInternalRequest } from './utils/esclient_with_internal_request';
 
 async function retryUntilAvailable<T>(
   func: () => Promise<T>,
@@ -151,10 +152,8 @@ async function initCodeNode(server: Server, serverOptions: ServerOptions, log: L
   const queueIndex: string = server.config().get('xpack.code.queueIndex');
   const queueTimeout: number = server.config().get('xpack.code.queueTimeout');
   const devMode: boolean = server.config().get('env.dev');
-  const adminCluster = server.plugins.elasticsearch.getCluster('admin');
 
-  // @ts-ignore
-  const esClient: EsClient = adminCluster.clusterClient.client;
+  const esClient: EsClient = new EsClientWithInternalRequest(server);
   const repoConfigController = new RepositoryConfigController(esClient);
 
   server.injectUiAppVars('code', () => ({

x-pack/plugins/code/server/lib/esqueue/misc.d.ts

@@ -33,6 +33,7 @@ export interface EsClient {
   search(params: AnyObject): Promise<any>;
   delete(params: AnyObject): Promise<any>;
   deleteByQuery(params: AnyObject): Promise<any>;
+  updateByQuery(params: AnyObject): Promise<any>;
 }
 
 export type LogFn = (msg: string | Error, tags: string[]) => void;

x-pack/plugins/code/server/utils/es_index_client.ts

@@ -6,9 +6,10 @@
 
 import { AnyObject } from '../lib/esqueue';
 import { WithRequest } from './with_request';
+import { WithInternalRequest } from './with_internal_request';
 
 export class EsIndexClient {
-  constructor(readonly self: WithRequest) {}
+  constructor(readonly self: WithRequest | WithInternalRequest) {}
 
   public exists(params: AnyObject): Promise<any> {
     return this.self.callCluster('indices.exists', params);

x-pack/plugins/code/server/utils/esclient_with_internal_request.ts

@@ -0,0 +1,58 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { Server } from 'hapi';
+import { AnyObject, EsClient } from '../lib/esqueue';
+import { EsIndexClient } from './es_index_client';
+import { WithInternalRequest } from './with_internal_request';
+
+export class EsClientWithInternalRequest extends WithInternalRequest implements EsClient {
+  public readonly indices = new EsIndexClient(this);
+
+  constructor(server: Server) {
+    super(server);
+  }
+
+  public bulk(params: AnyObject): Promise<any> {
+    return this.callCluster('bulk', params);
+  }
+
+  public delete(params: AnyObject): Promise<any> {
+    return this.callCluster('delete', params);
+  }
+
+  public deleteByQuery(params: AnyObject): Promise<any> {
+    return this.callCluster('deleteByQuery', params);
+  }
+
+  public get(params: AnyObject): Promise<any> {
+    return this.callCluster('get', params);
+  }
+
+  public index(params: AnyObject): Promise<any> {
+    return this.callCluster('index', params);
+  }
+
+  public ping(): Promise<void> {
+    return this.callCluster('ping');
+  }
+
+  public reindex(params: AnyObject): Promise<any> {
+    return this.callCluster('reindex', params);
+  }
+
+  public search(params: AnyObject): Promise<any> {
+    return this.callCluster('search', params);
+  }
+
+  public update(params: AnyObject): Promise<any> {
+    return this.callCluster('update', params);
+  }
+
+  public updateByQuery(params: AnyObject): Promise<any> {
+    return this.callCluster('updateByQuery', params);
+  }
+}

x-pack/plugins/code/server/utils/esclient_with_request.ts

@@ -51,4 +51,8 @@ export class EsClientWithRequest extends WithRequest implements EsClient {
   public update(params: AnyObject): Promise<any> {
     return this.callCluster('update', params);
   }
+
+  public updateByQuery(params: AnyObject): Promise<any> {
+    return this.callCluster('updateByQuery', params);
+  }
 }

x-pack/plugins/code/server/utils/with_internal_request.ts

@@ -0,0 +1,17 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { Server } from 'hapi';
+import { AnyObject } from '../lib/esqueue';
+
+export class WithInternalRequest {
+  public readonly callCluster: (endpoint: string, clientOptions?: AnyObject) => Promise<any>;
+
+  constructor(server: Server) {
+    const cluster = server.plugins.elasticsearch.getCluster('admin');
+    this.callCluster = cluster.callWithInternalUser;
+  }
+}