Update logstash pipeline management to use system index APIs (#80405)

This change updates the logstash pipeline management plugin to use
pipeline management APIs in Elasticsearch rather than directly
accessing the .logstash index. In Elasticsearch 8.0, direct access to
system indices will no longer be allowed when using standard APIs.
Given this change, a new set of APIs has been created specifically for
the management of Logstash pipelines and this change makes use of the
APIs.

Co-authored-by: Kaise Cheng <kaise.cheng@elastic.co>
Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

ES:#53350
LS:#12291

x-pack/plugins/logstash/server/models/pipeline/pipeline.test.ts

@@ -10,8 +10,7 @@ describe('pipeline', () => {
   describe('Pipeline', () => {
     describe('fromUpstreamJSON factory method', () => {
       const upstreamJSON = {
-        _id: 'apache',
-        _source: {
+        apache: {
           description: 'this is an apache pipeline',
           pipeline_metadata: {
             version: 1,
@@ -21,25 +20,23 @@ describe('pipeline', () => {
           pipeline: 'input {} filter { grok {} }\n output {}',
         },
       };
+      const upstreamId = 'apache';
 
       it('returns correct Pipeline instance', () => {
         const pipeline = Pipeline.fromUpstreamJSON(upstreamJSON);
-        expect(pipeline.id).toBe(upstreamJSON._id);
-        expect(pipeline.description).toBe(upstreamJSON._source.description);
-        expect(pipeline.username).toBe(upstreamJSON._source.username);
-        expect(pipeline.pipeline).toBe(upstreamJSON._source.pipeline);
+        expect(pipeline.id).toBe(upstreamId);
+        expect(pipeline.description).toBe(upstreamJSON.apache.description);
+        expect(pipeline.username).toBe(upstreamJSON.apache.username);
+        expect(pipeline.pipeline).toBe(upstreamJSON.apache.pipeline);
       });
 
-      it('throws if pipeline argument does not contain an id property', () => {
-        const badJSON = {
-          // no _id
-          _source: upstreamJSON._source,
-        };
+      it('throws if pipeline argument does not contain id as a key', () => {
+        const badJSON = {};
         const testFromUpstreamJsonError = () => {
           return Pipeline.fromUpstreamJSON(badJSON);
         };
         expect(testFromUpstreamJsonError).toThrowError(
-          /upstreamPipeline argument must contain an id property/i
+          /upstreamPipeline argument must contain pipeline id as a key/i
         );
       });
     });

x-pack/plugins/logstash/server/models/pipeline/pipeline.ts

@@ -93,21 +93,21 @@ export class Pipeline {
 
   // generate Pipeline object from elasticsearch response
   static fromUpstreamJSON(upstreamPipeline: Record<string, any>) {
-    if (!upstreamPipeline._id) {
+    if (Object.keys(upstreamPipeline).length !== 1) {
       throw badRequest(
         i18n.translate(
           'xpack.logstash.upstreamPipelineArgumentMustContainAnIdPropertyErrorMessage',
           {
-            defaultMessage: 'upstreamPipeline argument must contain an id property',
+            defaultMessage: 'upstreamPipeline argument must contain pipeline id as a key',
           }
         )
       );
     }
-    const id = get(upstreamPipeline, '_id') as string;
-    const description = get(upstreamPipeline, '_source.description') as string;
-    const username = get(upstreamPipeline, '_source.username') as string;
-    const pipeline = get(upstreamPipeline, '_source.pipeline') as string;
-    const settings = get(upstreamPipeline, '_source.pipeline_settings') as Record<string, any>;
+    const id = Object.keys(upstreamPipeline).pop() as string;
+    const description = get(upstreamPipeline, id + '.description') as string;
+    const username = get(upstreamPipeline, id + '.username') as string;
+    const pipeline = get(upstreamPipeline, id + '.pipeline') as string;
+    const settings = get(upstreamPipeline, id + '.pipeline_settings') as Record<string, any>;
 
     const opts: PipelineOptions = { id, description, username, pipeline, settings };
 

x-pack/plugins/logstash/server/models/pipeline_list_item/pipeline_list_item.test.ts

@@ -9,8 +9,7 @@ import { PipelineListItem } from './pipeline_list_item';
 describe('pipeline_list_item', () => {
   describe('PipelineListItem', () => {
     const upstreamJSON = {
-      _id: 'apache',
-      _source: {
+      apache: {
         description: 'this is an apache pipeline',
         last_modified: '2017-05-14T02:50:51.250Z',
         pipeline_metadata: {
@@ -20,24 +19,22 @@ describe('pipeline_list_item', () => {
         username: 'elastic',
         pipeline: 'input {} filter { grok {} }\n output {}',
       },
-      _index: 'index',
-      _type: 'type',
-      _score: 100,
     };
+    const upstreamId = 'apache';
 
     describe('fromUpstreamJSON factory method', () => {
       it('returns correct PipelineListItem instance', () => {
-        const pipelineListItem = PipelineListItem.fromUpstreamJSON(upstreamJSON);
-        expect(pipelineListItem.id).toBe(upstreamJSON._id);
-        expect(pipelineListItem.description).toBe(upstreamJSON._source.description);
-        expect(pipelineListItem.username).toBe(upstreamJSON._source.username);
-        expect(pipelineListItem.last_modified).toBe(upstreamJSON._source.last_modified);
+        const pipelineListItem = PipelineListItem.fromUpstreamJSON(upstreamId, upstreamJSON);
+        expect(pipelineListItem.id).toBe(upstreamId);
+        expect(pipelineListItem.description).toBe(upstreamJSON.apache.description);
+        expect(pipelineListItem.username).toBe(upstreamJSON.apache.username);
+        expect(pipelineListItem.last_modified).toBe(upstreamJSON.apache.last_modified);
       });
     });
 
     describe('downstreamJSON getter method', () => {
       it('returns the downstreamJSON JSON', () => {
-        const pipelineListItem = PipelineListItem.fromUpstreamJSON(upstreamJSON);
+        const pipelineListItem = PipelineListItem.fromUpstreamJSON(upstreamId, upstreamJSON);
         const expectedDownstreamJSON = {
           id: 'apache',
           description: 'this is an apache pipeline',

x-pack/plugins/logstash/server/models/pipeline_list_item/pipeline_list_item.ts

@@ -5,7 +5,7 @@
  */
 
 import { get } from 'lodash';
-import { Hit, PipelineListItemOptions } from '../../types';
+import { PipelineListItemOptions } from '../../types';
 
 export class PipelineListItem {
   public readonly id: string;
@@ -34,12 +34,12 @@ export class PipelineListItem {
    * Takes the json GET response from ES and constructs a pipeline model to be used
    * in Kibana downstream
    */
-  static fromUpstreamJSON(pipeline: Hit) {
+  static fromUpstreamJSON(id: string, pipeline: Record<string, any>) {
     const opts = {
-      id: pipeline._id,
-      description: get(pipeline, '_source.description') as string,
-      last_modified: get(pipeline, '_source.last_modified') as string,
-      username: get(pipeline, '_source.username') as string,
+      id,
+      description: get(pipeline, id + '.description') as string,
+      last_modified: get(pipeline, id + '.last_modified') as string,
+      username: get(pipeline, id + '.username') as string,
     };
 
     return new PipelineListItem(opts);

x-pack/plugins/logstash/server/plugin.ts

@@ -44,10 +44,8 @@ export class LogstashPlugin implements Plugin {
       },
       privileges: [
         {
-          requiredClusterPrivileges: [],
-          requiredIndexPrivileges: {
-            ['.logstash']: ['read'],
-          },
+          requiredClusterPrivileges: ['manage_logstash_pipelines'],
+          requiredIndexPrivileges: {},
           ui: [],
         },
       ],

x-pack/plugins/logstash/server/routes/pipeline/delete.ts

@@ -5,7 +5,6 @@
  */
 import { schema } from '@kbn/config-schema';
 import { IRouter } from 'src/core/server';
-import { INDEX_NAMES } from '../../../common/constants';
 import { wrapRouteWithLicenseCheck } from '../../../../licensing/server';
 
 import { checkLicense } from '../../lib/check_license';
@@ -25,10 +24,9 @@ export function registerPipelineDeleteRoute(router: IRouter) {
       router.handleLegacyErrors(async (context, request, response) => {
         const client = context.logstash!.esClient;
 
-        await client.callAsCurrentUser('delete', {
-          index: INDEX_NAMES.PIPELINES,
-          id: request.params.id,
-          refresh: 'wait_for',
+        await client.callAsCurrentUser('transport.request', {
+          path: '/_logstash/pipeline/' + encodeURIComponent(request.params.id),
+          method: 'DELETE',
         });
 
         return response.noContent();

x-pack/plugins/logstash/server/routes/pipeline/load.ts

@@ -6,7 +6,6 @@
 import { schema } from '@kbn/config-schema';
 import { IRouter } from 'src/core/server';
 
-import { INDEX_NAMES } from '../../../common/constants';
 import { Pipeline } from '../../models/pipeline';
 import { wrapRouteWithLicenseCheck } from '../../../../licensing/server';
 import { checkLicense } from '../../lib/check_license';
@@ -26,14 +25,13 @@ export function registerPipelineLoadRoute(router: IRouter) {
       router.handleLegacyErrors(async (context, request, response) => {
         const client = context.logstash!.esClient;
 
-        const result = await client.callAsCurrentUser('get', {
-          index: INDEX_NAMES.PIPELINES,
-          id: request.params.id,
-          _source: ['description', 'username', 'pipeline', 'pipeline_settings'],
+        const result = await client.callAsCurrentUser('transport.request', {
+          path: '/_logstash/pipeline/' + encodeURIComponent(request.params.id),
+          method: 'GET',
           ignore: [404],
         });
 
-        if (!result.found) {
+        if (result[request.params.id] === undefined) {
           return response.notFound();
         }
 

x-pack/plugins/logstash/server/routes/pipeline/save.ts

@@ -7,7 +7,6 @@ import { schema } from '@kbn/config-schema';
 import { i18n } from '@kbn/i18n';
 import { IRouter } from 'src/core/server';
 
-import { INDEX_NAMES } from '../../../common/constants';
 import { Pipeline } from '../../models/pipeline';
 import { wrapRouteWithLicenseCheck } from '../../../../licensing/server';
 import { SecurityPluginSetup } from '../../../../security/server';
@@ -41,11 +40,10 @@ export function registerPipelineSaveRoute(router: IRouter, security?: SecurityPl
           const client = context.logstash!.esClient;
           const pipeline = Pipeline.fromDownstreamJSON(request.body, request.params.id, username);
 
-          await client.callAsCurrentUser('index', {
-            index: INDEX_NAMES.PIPELINES,
-            id: pipeline.id,
+          await client.callAsCurrentUser('transport.request', {
+            path: '/_logstash/pipeline/' + encodeURIComponent(pipeline.id),
+            method: 'PUT',
             body: pipeline.upstreamJSON,
-            refresh: 'wait_for',
           });
 
           return response.noContent();

x-pack/plugins/logstash/server/routes/pipelines/delete.ts

@@ -7,15 +7,13 @@ import { schema } from '@kbn/config-schema';
 import { LegacyAPICaller, IRouter } from 'src/core/server';
 import { wrapRouteWithLicenseCheck } from '../../../../licensing/server';
 
-import { INDEX_NAMES } from '../../../common/constants';
 import { checkLicense } from '../../lib/check_license';
 
 async function deletePipelines(callWithRequest: LegacyAPICaller, pipelineIds: string[]) {
   const deletePromises = pipelineIds.map((pipelineId) => {
-    return callWithRequest('delete', {
-      index: INDEX_NAMES.PIPELINES,
-      id: pipelineId,
-      refresh: 'wait_for',
+    return callWithRequest('transport.request', {
+      path: '/_logstash/pipeline/' + encodeURIComponent(pipelineId),
+      method: 'DELETE',
     })
       .then((success) => ({ success }))
       .catch((error) => ({ error }));