[7.x] Update logstash pipeline management to use system index APIs (#80405)

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 }));