Skip to content

Commit

Permalink
Update logstash pipeline management to use system index APIs (#80405) (
Browse files Browse the repository at this point in the history
…#83526)

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 <[email protected]>
Co-authored-by: Kibana Machine <[email protected]>

ES:#53350
LS:#12291

Co-authored-by: Jay Modi <[email protected]>
  • Loading branch information
kaisecheng and jaymode authored Nov 17, 2020
1 parent 7357680 commit a338220
Show file tree
Hide file tree
Showing 17 changed files with 93 additions and 238 deletions.

This file was deleted.

This file was deleted.

This file was deleted.

21 changes: 9 additions & 12 deletions x-pack/plugins/logstash/server/models/pipeline/pipeline.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
);
});
});
Expand Down
14 changes: 7 additions & 7 deletions x-pack/plugins/logstash/server/models/pipeline/pipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 };

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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: {
Expand All @@ -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',
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
*/

import { get } from 'lodash';
import { Hit, PipelineListItemOptions } from '../../types';
import { PipelineListItemOptions } from '../../types';

export class PipelineListItem {
public readonly id: string;
Expand Down Expand Up @@ -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);
Expand Down
6 changes: 2 additions & 4 deletions x-pack/plugins/logstash/server/plugin.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,8 @@ export class LogstashPlugin implements Plugin {
},
privileges: [
{
requiredClusterPrivileges: [],
requiredIndexPrivileges: {
['.logstash']: ['read'],
},
requiredClusterPrivileges: ['manage_logstash_pipelines'],
requiredIndexPrivileges: {},
ui: [],
},
],
Expand Down
8 changes: 3 additions & 5 deletions x-pack/plugins/logstash/server/routes/pipeline/delete.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -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();
Expand Down
10 changes: 4 additions & 6 deletions x-pack/plugins/logstash/server/routes/pipeline/load.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -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();
}

Expand Down
8 changes: 3 additions & 5 deletions x-pack/plugins/logstash/server/routes/pipeline/save.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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();
Expand Down
8 changes: 3 additions & 5 deletions x-pack/plugins/logstash/server/routes/pipelines/delete.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 }));
Expand Down
Loading

0 comments on commit a338220

Please sign in to comment.