Skip to content

Commit

Permalink
Refactor es reference updating
Browse files Browse the repository at this point in the history
  • Loading branch information
@joshdover
joshdover committed May 2, 2022
1 parent f275d45 commit d22c8dd
Show file tree
Hide file tree
Showing 15 changed files with 311 additions and 222 deletions.
64 changes: 34 additions & 30 deletions x-pack/plugins/fleet/server/services/epm/elasticsearch/datastream_ilm/install.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,13 @@ import type {
InstallablePackage,
RegistryDataStream,
} from '../../../../../common/types/models';
import { getInstallation } from '../../packages';
import { saveInstalledEsRefs } from '../../packages/install';
import { updateEsAssetReferences } from '../../packages/install';
import { getAsset } from '../transform/common';

import { getESAssetMetadata } from '../meta';
import { retryTransientEsErrors } from '../retry';

import { deleteIlmRefs, deleteIlms } from './remove';
import { deleteIlms } from './remove';

interface IlmInstallation {
installationName: string;
Expand All @@ -37,24 +36,39 @@ export const installIlmForDataStream = async (
paths: string[],
esClient: ElasticsearchClient,
savedObjectsClient: SavedObjectsClientContract,
logger: Logger
logger: Logger,
esReferences: EsAssetReference[]
) => {
const installation = await getInstallation({ savedObjectsClient, pkgName: registryPackage.name });
let previousInstalledIlmEsAssets: EsAssetReference[] = [];
if (installation) {
previousInstalledIlmEsAssets = installation.installed_es.filter(
({ type, id }) => type === ElasticsearchAssetType.dataStreamIlmPolicy
);
}
const previousInstalledIlmEsAssets = esReferences.filter(
({ type }) => type === ElasticsearchAssetType.dataStreamIlmPolicy
);

// delete all previous ilm
await deleteIlms(
esClient,
previousInstalledIlmEsAssets.map((asset) => asset.id)
);

if (previousInstalledIlmEsAssets.length > 0) {
// remove the saved object reference
esReferences = await updateEsAssetReferences(
savedObjectsClient,
registryPackage.name,
esReferences,
{
assetsToRemove: previousInstalledIlmEsAssets,
}
);
}

// install the latest dataset
const dataStreams = registryPackage.data_streams;
if (!dataStreams?.length) return [];
if (!dataStreams?.length)
return {
installedIlms: [],
esReferences,
};

const dataStreamIlmPaths = paths.filter((path) => isDataStreamIlm(path));
let installedIlms: EsAssetReference[] = [];
if (dataStreamIlmPaths.length > 0) {
Expand All @@ -77,12 +91,17 @@ export const installIlmForDataStream = async (
return acc;
}, []);

await saveInstalledEsRefs(savedObjectsClient, registryPackage.name, ilmRefs);
esReferences = await updateEsAssetReferences(
savedObjectsClient,
registryPackage.name,
esReferences,
{ assetsToAdd: ilmRefs }
);

const ilmInstallations: IlmInstallation[] = ilmPathDatasets.map(
(ilmPathDataset: IlmPathDataset) => {
const content = JSON.parse(getAsset(ilmPathDataset.path).toString('utf-8'));
content.policy._meta = getESAssetMetadata({ packageName: installation?.name });
content.policy._meta = getESAssetMetadata({ packageName: registryPackage.name });

return {
installationName: getIlmNameForInstallation(ilmPathDataset),
Expand All @@ -98,22 +117,7 @@ export const installIlmForDataStream = async (
installedIlms = await Promise.all(installationPromises).then((results) => results.flat());
}

if (previousInstalledIlmEsAssets.length > 0) {
const currentInstallation = await getInstallation({
savedObjectsClient,
pkgName: registryPackage.name,
});

// remove the saved object reference
await deleteIlmRefs(
savedObjectsClient,
currentInstallation?.installed_es || [],
registryPackage.name,
previousInstalledIlmEsAssets.map((asset) => asset.id),
installedIlms.map((installed) => installed.id)
);
}
return installedIlms;
return { installedIlms, esReferences };
};

async function handleIlmInstall({
Expand Down
27 changes: 1 addition & 26 deletions x-pack/plugins/fleet/server/services/epm/elasticsearch/datastream_ilm/remove.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,7 @@
* 2.0.
*/

import type { ElasticsearchClient, SavedObjectsClientContract } from '@kbn/core/server';

import { ElasticsearchAssetType } from '../../../../types';
import type { EsAssetReference } from '../../../../types';
import { PACKAGES_SAVED_OBJECT_TYPE } from '../../../../../common/constants';
import type { ElasticsearchClient } from '@kbn/core/server';

export const deleteIlms = async (esClient: ElasticsearchClient, ilmPolicyIds: string[]) => {
await Promise.all(
Expand All @@ -26,24 +22,3 @@ export const deleteIlms = async (esClient: ElasticsearchClient, ilmPolicyIds: st
})
);
};

export const deleteIlmRefs = async (
savedObjectsClient: SavedObjectsClientContract,
installedEsAssets: EsAssetReference[],
pkgName: string,
installedEsIdToRemove: string[],
currentInstalledEsIlmIds: string[]
) => {
const seen = new Set<string>();
const filteredAssets = installedEsAssets.filter(({ type, id }) => {
if (type !== ElasticsearchAssetType.dataStreamIlmPolicy) return true;
const add =
(currentInstalledEsIlmIds.includes(id) || !installedEsIdToRemove.includes(id)) &&
!seen.has(id);
seen.add(id);
return add;
});
return savedObjectsClient.update(PACKAGES_SAVED_OBJECT_TYPE, pkgName, {
installed_es: filteredAssets,
});
};
44 changes: 31 additions & 13 deletions x-pack/plugins/fleet/server/services/epm/elasticsearch/ilm/install.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,38 +5,54 @@
* 2.0.
*/

import type { ElasticsearchClient, Logger } from '@kbn/core/server';
import type { ElasticsearchClient, Logger, SavedObjectsClientContract } from '@kbn/core/server';

import type { InstallablePackage } from '../../../../types';
import type { EsAssetReference, InstallablePackage } from '../../../../types';

import { ElasticsearchAssetType } from '../../../../types';
import { getAsset, getPathParts } from '../../archive';
import { updateEsAssetReferences } from '../../packages/install';
import { getESAssetMetadata } from '../meta';
import { retryTransientEsErrors } from '../retry';

export async function installILMPolicy(
packageInfo: InstallablePackage,
paths: string[],
esClient: ElasticsearchClient,
logger: Logger
) {
savedObjectsClient: SavedObjectsClientContract,
logger: Logger,
esReferences: EsAssetReference[]
): Promise<EsAssetReference[]> {
const ilmPaths = paths.filter((path) => isILMPolicy(path));
if (!ilmPaths.length) return;
await Promise.all(
ilmPaths.map(async (path) => {
const body = JSON.parse(getAsset(path).toString('utf-8'));
if (!ilmPaths.length) return esReferences;

const ilmPolicies = ilmPaths.map((path) => {
const body = JSON.parse(getAsset(path).toString('utf-8'));

body.policy._meta = getESAssetMetadata({ packageName: packageInfo.name });

const { file } = getPathParts(path);
const name = file.substr(0, file.lastIndexOf('.'));

body.policy._meta = getESAssetMetadata({ packageName: packageInfo.name });
return { name, body };
});

const { file } = getPathParts(path);
const name = file.substr(0, file.lastIndexOf('.'));
esReferences = await updateEsAssetReferences(savedObjectsClient, packageInfo.name, esReferences, {
assetsToAdd: ilmPolicies.map((policy) => ({
type: ElasticsearchAssetType.ilmPolicy,
id: policy.name,
})),
});

await Promise.all(
ilmPolicies.map(async (policy) => {
try {
await retryTransientEsErrors(
() =>
esClient.transport.request({
method: 'PUT',
path: '/_ilm/policy/' + name,
body,
path: '/_ilm/policy/' + policy.name,
body: policy.body,
}),
{ logger }
);
Expand All @@ -45,6 +61,8 @@ export async function installILMPolicy(
}
})
);

return esReferences;
}

const isILMPolicy = (path: string) => {
Expand Down
34 changes: 13 additions & 21 deletions x-pack/plugins/fleet/server/services/epm/elasticsearch/ingest_pipeline/install.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,7 @@ import { ElasticsearchAssetType } from '../../../../types';
import type { EsAssetReference, RegistryDataStream, InstallablePackage } from '../../../../types';
import { getAsset, getPathParts } from '../../archive';
import type { ArchiveEntry } from '../../archive';
import { saveInstalledEsRefs } from '../../packages/install';
import { getInstallationObject } from '../../packages';
import { updateEsAssetReferences } from '../../packages/install';
import {
FLEET_FINAL_PIPELINE_CONTENT,
FLEET_FINAL_PIPELINE_ID,
Expand All @@ -24,8 +23,6 @@ import { appendMetadataToIngestPipeline } from '../meta';

import { retryTransientEsErrors } from '../retry';

import { deletePipelineRefs } from './remove';

interface RewriteSubstitution {
source: string;
target: string;
Expand All @@ -44,7 +41,8 @@ export const installPipelines = async (
paths: string[],
esClient: ElasticsearchClient,
savedObjectsClient: SavedObjectsClientContract,
logger: Logger
logger: Logger,
esReferences: EsAssetReference[]
) => {
// unlike other ES assets, pipeline names are versioned so after a template is updated
// it can be created pointing to the new template, without removing the old one and effecting data
Expand All @@ -67,7 +65,7 @@ export const installPipelines = async (
const nameForInstallation = getPipelineNameForInstallation({
pipelineName: name,
dataStream,
packageVersion: installablePackage.version,
packageVersion: pkgVersion,
});
return { id: nameForInstallation, type: ElasticsearchAssetType.ingestPipeline };
});
Expand All @@ -80,27 +78,20 @@ export const installPipelines = async (
const { name } = getNameAndExtension(path);
const nameForInstallation = getPipelineNameForInstallation({
pipelineName: name,
packageVersion: installablePackage.version,
packageVersion: pkgVersion,
});
return { id: nameForInstallation, type: ElasticsearchAssetType.ingestPipeline };
});

pipelineRefs = [...pipelineRefs, ...topLevelPipelineRefs];

// check that we don't duplicate the pipeline refs if the user is reinstalling
const installedPkg = await getInstallationObject({
savedObjectsClient,
pkgName,
esReferences = await updateEsAssetReferences(savedObjectsClient, pkgName, esReferences, {
assetsToRemove: esReferences.filter(
(ref) => ref.type === ElasticsearchAssetType.ingestPipeline && !ref.id.includes(pkgVersion)
),
assetsToAdd: pipelineRefs,
});
if (!installedPkg) throw new Error("integration wasn't found while installing pipelines");
// remove the current pipeline refs, if any exist, associated with this version before saving new ones so no duplicates occur
await deletePipelineRefs(
savedObjectsClient,
installedPkg.attributes.installed_es,
pkgName,
pkgVersion
);
await saveInstalledEsRefs(savedObjectsClient, installablePackage.name, pipelineRefs);

const pipelines = dataStreams
? dataStreams.reduce<Array<Promise<EsAssetReference[]>>>((acc, dataStream) => {
if (dataStream.ingest_pipeline) {
Expand Down Expand Up @@ -130,7 +121,8 @@ export const installPipelines = async (
);
}

return await Promise.all(pipelines).then((results) => results.flat());
await Promise.all(pipelines);
return esReferences;
};

export function rewriteIngestPipeline(
Expand Down
22 changes: 9 additions & 13 deletions x-pack/plugins/fleet/server/services/epm/elasticsearch/ingest_pipeline/remove.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,8 @@ import { appContextService } from '../../..';
import { ElasticsearchAssetType } from '../../../../types';
import { IngestManagerError } from '../../../../errors';
import { getInstallation } from '../../packages/get';
import { PACKAGES_SAVED_OBJECT_TYPE } from '../../../../../common';
import type { EsAssetReference } from '../../../../../common';
import { updateEsAssetReferences } from '../../packages/install';

export const deletePreviousPipelines = async (
esClient: ElasticsearchClient,
Expand Down Expand Up @@ -49,20 +49,16 @@ export const deletePipelineRefs = async (
pkgName: string,
pkgVersion: string
) => {
const filteredAssets = installedEsAssets.filter(({ type, id }) => {
if (type !== ElasticsearchAssetType.ingestPipeline) return true;
if (!id.includes(pkgVersion)) return true;
return false;
const assetsToRemove = installedEsAssets.filter(({ type, id }) => {
return type === ElasticsearchAssetType.ingestPipeline && id.includes(pkgVersion);
});

return updateEsAssetReferences(savedObjectsClient, pkgName, installedEsAssets, {
assetsToRemove,
refresh: 'wait_for',
});
return savedObjectsClient.update(
PACKAGES_SAVED_OBJECT_TYPE,
pkgName,
{
installed_es: filteredAssets,
},
{ refresh: false }
);
};

export async function deletePipeline(esClient: ElasticsearchClient, id: string): Promise<void> {
// '*' shouldn't ever appear here, but it still would delete all ingest pipelines
if (id && id !== '*') {
Expand Down
Loading

0 comments on commit d22c8dd

Please sign in to comment.