Skip to content

Commit

Permalink
[7.9] [Saved objects] Add support for version on create & bulkCreate …
Browse files Browse the repository at this point in the history
…when overwriting a document (elastic#75172) (elastic#75498)

* [Saved objects] Add support for version on create & bulkCreate when overwriting a document (elastic#75172)

Adds support for `version` one the SavedObjectsClient's create api.
This sallows us to retain Optimistic concurrency control when using create to overwrite an existing document.
  • Loading branch information
gmmorris authored Aug 20, 2020
1 parent 4e6cecb commit 00836e5
Show file tree
Hide file tree
Showing 12 changed files with 104 additions and 10 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -20,4 +20,5 @@ export interface SavedObjectsBulkCreateObject<T = unknown>
| [migrationVersion](./kibana-plugin-core-server.savedobjectsbulkcreateobject.migrationversion.md) | <code>SavedObjectsMigrationVersion</code> | Information about the migrations that have been applied to this SavedObject. When Kibana starts up, KibanaMigrator detects outdated documents and migrates them based on this value. For each migration that has been applied, the plugin's name is used as a key and the latest migration version as the value. |
| [references](./kibana-plugin-core-server.savedobjectsbulkcreateobject.references.md) | <code>SavedObjectReference[]</code> | |
| [type](./kibana-plugin-core-server.savedobjectsbulkcreateobject.type.md) | <code>string</code> | |
| [version](./kibana-plugin-core-server.savedobjectsbulkcreateobject.version.md) | <code>string</code> | |

Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
<!-- Do not edit this file. It is automatically generated by API Documenter. -->

[Home](./index.md) &gt; [kibana-plugin-core-server](./kibana-plugin-core-server.md) &gt; [SavedObjectsBulkCreateObject](./kibana-plugin-core-server.savedobjectsbulkcreateobject.md) &gt; [version](./kibana-plugin-core-server.savedobjectsbulkcreateobject.version.md)

## SavedObjectsBulkCreateObject.version property

<b>Signature:</b>

```typescript
version?: string;
```
Original file line number Diff line number Diff line change
Expand Up @@ -20,4 +20,5 @@ export interface SavedObjectsCreateOptions extends SavedObjectsBaseOptions
| [overwrite](./kibana-plugin-core-server.savedobjectscreateoptions.overwrite.md) | <code>boolean</code> | Overwrite existing documents (defaults to false) |
| [references](./kibana-plugin-core-server.savedobjectscreateoptions.references.md) | <code>SavedObjectReference[]</code> | |
| [refresh](./kibana-plugin-core-server.savedobjectscreateoptions.refresh.md) | <code>MutatingOperationRefreshSetting</code> | The Elasticsearch Refresh setting for this operation |
| [version](./kibana-plugin-core-server.savedobjectscreateoptions.version.md) | <code>string</code> | An opaque version number which changes on each successful write operation. Can be used in conjunction with <code>overwrite</code> for implementing optimistic concurrency control. |
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
<!-- Do not edit this file. It is automatically generated by API Documenter. -->

[Home](./index.md) &gt; [kibana-plugin-core-server](./kibana-plugin-core-server.md) &gt; [SavedObjectsCreateOptions](./kibana-plugin-core-server.savedobjectscreateoptions.md) &gt; [version](./kibana-plugin-core-server.savedobjectscreateoptions.version.md)

## SavedObjectsCreateOptions.version property

An opaque version number which changes on each successful write operation. Can be used in conjunction with `overwrite` for implementing optimistic concurrency control.

<b>Signature:</b>

```typescript
version?: string;
```
7 changes: 6 additions & 1 deletion src/core/server/saved_objects/import/import_saved_objects.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import {
SavedObjectsImportOptions,
} from './types';
import { validateReferences } from './validate_references';
import { SavedObject } from '../types';

/**
* Import saved objects from given stream. See the {@link SavedObjectsImportOptions | options} for more
Expand Down Expand Up @@ -67,7 +68,7 @@ export async function importSavedObjectsFromStream({
}

// Create objects in bulk
const bulkCreateResult = await savedObjectsClient.bulkCreate(filteredObjects, {
const bulkCreateResult = await savedObjectsClient.bulkCreate(omitVersion(filteredObjects), {
overwrite,
namespace,
});
Expand All @@ -82,3 +83,7 @@ export async function importSavedObjectsFromStream({
...(errorAccumulator.length ? { errors: errorAccumulator } : {}),
};
}

export function omitVersion(objects: SavedObject[]): SavedObject[] {
return objects.map(({ version, ...object }) => object);
}
12 changes: 8 additions & 4 deletions src/core/server/saved_objects/import/resolve_import_errors.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import {
SavedObjectsResolveImportErrorsOptions,
} from './types';
import { validateReferences } from './validate_references';
import { omitVersion } from './import_saved_objects';

/**
* Resolve and return saved object import errors.
Expand Down Expand Up @@ -91,7 +92,7 @@ export async function resolveSavedObjectsImportErrors({
// Bulk create in two batches, overwrites and non-overwrites
const { objectsToOverwrite, objectsToNotOverwrite } = splitOverwrites(filteredObjects, retries);
if (objectsToOverwrite.length) {
const bulkCreateResult = await savedObjectsClient.bulkCreate(objectsToOverwrite, {
const bulkCreateResult = await savedObjectsClient.bulkCreate(omitVersion(objectsToOverwrite), {
overwrite: true,
namespace,
});
Expand All @@ -102,9 +103,12 @@ export async function resolveSavedObjectsImportErrors({
successCount += bulkCreateResult.saved_objects.filter((obj) => !obj.error).length;
}
if (objectsToNotOverwrite.length) {
const bulkCreateResult = await savedObjectsClient.bulkCreate(objectsToNotOverwrite, {
namespace,
});
const bulkCreateResult = await savedObjectsClient.bulkCreate(
omitVersion(objectsToNotOverwrite),
{
namespace,
}
);
errorAccumulator = [
...errorAccumulator,
...extractErrors(bulkCreateResult.saved_objects, objectsToNotOverwrite),
Expand Down
42 changes: 40 additions & 2 deletions src/core/server/saved_objects/service/lib/repository.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -472,8 +472,16 @@ describe('SavedObjectsRepository', () => {
{ method, _index = expect.any(String), getId = () => expect.any(String) }
) => {
const body = [];
for (const { type, id } of objects) {
body.push({ [method]: { _index, _id: getId(type, id) } });
for (const { type, id, if_primary_term: ifPrimaryTerm, if_seq_no: ifSeqNo } of objects) {
body.push({
[method]: {
_index,
_id: getId(type, id),
...(ifPrimaryTerm && ifSeqNo
? { if_primary_term: expect.any(Number), if_seq_no: expect.any(Number) }
: {}),
},
});
body.push(expect.any(Object));
}
expectClusterCallArgs({ body });
Expand Down Expand Up @@ -529,6 +537,27 @@ describe('SavedObjectsRepository', () => {
expectClusterCallArgsAction([obj1, obj2], { method: 'index' });
});

it(`should use the ES index method with version if ID and version are defined and overwrite=true`, async () => {
await bulkCreateSuccess(
[
{
...obj1,
version: mockVersion,
},
obj2,
],
{ overwrite: true }
);

const obj1WithSeq = {
...obj1,
if_seq_no: mockVersionProps._seq_no,
if_primary_term: mockVersionProps._primary_term,
};

expectClusterCallArgsAction([obj1WithSeq, obj2], { method: 'index' });
});

it(`should use the ES create method if ID is defined and overwrite=false`, async () => {
await bulkCreateSuccess([obj1, obj2]);
expectClusterCallArgsAction([obj1, obj2], { method: 'create' });
Expand Down Expand Up @@ -1458,6 +1487,15 @@ describe('SavedObjectsRepository', () => {
expectClusterCalls('index');
});

it(`should use the ES index with version if ID and version are defined and overwrite=true`, async () => {
await createSuccess(type, attributes, { id, overwrite: true, version: mockVersion });
expectClusterCalls('index');
expectClusterCallArgs({
if_seq_no: mockVersionProps._seq_no,
if_primary_term: mockVersionProps._primary_term,
});
});

it(`should use the ES create action if ID is defined and overwrite=false`, async () => {
await createSuccess(type, attributes, { id });
expectClusterCalls('create');
Expand Down
12 changes: 11 additions & 1 deletion src/core/server/saved_objects/service/lib/repository.ts
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,7 @@ export class SavedObjectsRepository {
overwrite = false,
references = [],
refresh = DEFAULT_REFRESH_SETTING,
version,
} = options;

if (!this._allowedTypes.includes(type)) {
Expand Down Expand Up @@ -260,6 +261,7 @@ export class SavedObjectsRepository {
index: this.getIndexForType(type),
refresh,
body: raw._source,
...(overwrite && version ? decodeRequestVersion(version) : {}),
});

return this._rawToSavedObject<T>({
Expand Down Expand Up @@ -339,7 +341,12 @@ export class SavedObjectsRepository {

let savedObjectNamespace;
let savedObjectNamespaces;
const { esRequestIndex, object, method } = expectedBulkGetResult.value;
let versionProperties;
const {
esRequestIndex,
object: { version, ...object },
method,
} = expectedBulkGetResult.value;
if (esRequestIndex !== undefined) {
const indexFound = bulkGetResponse.status !== 404;
const actualResult = indexFound ? bulkGetResponse.docs[esRequestIndex] : undefined;
Expand All @@ -356,12 +363,14 @@ export class SavedObjectsRepository {
};
}
savedObjectNamespaces = getSavedObjectNamespaces(namespace, docFound && actualResult);
versionProperties = getExpectedVersionProperties(version, actualResult);
} else {
if (this._registry.isSingleNamespace(object.type)) {
savedObjectNamespace = namespace;
} else if (this._registry.isMultiNamespace(object.type)) {
savedObjectNamespaces = getSavedObjectNamespaces(namespace);
}
versionProperties = getExpectedVersionProperties(version);
}

const expectedResult = {
Expand All @@ -386,6 +395,7 @@ export class SavedObjectsRepository {
[method]: {
_id: expectedResult.rawMigratedDoc._id,
_index: this.getIndexForType(object.type),
...(overwrite && versionProperties),
},
},
expectedResult.rawMigratedDoc._source
Expand Down
6 changes: 6 additions & 0 deletions src/core/server/saved_objects/service/saved_objects_client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,11 @@ export interface SavedObjectsCreateOptions extends SavedObjectsBaseOptions {
id?: string;
/** Overwrite existing documents (defaults to false) */
overwrite?: boolean;
/**
* An opaque version number which changes on each successful write operation.
* Can be used in conjunction with `overwrite` for implementing optimistic concurrency control.
**/
version?: string;
/** {@inheritDoc SavedObjectsMigrationVersion} */
migrationVersion?: SavedObjectsMigrationVersion;
references?: SavedObjectReference[];
Expand All @@ -52,6 +57,7 @@ export interface SavedObjectsBulkCreateObject<T = unknown> {
id?: string;
type: string;
attributes: T;
version?: string;
references?: SavedObjectReference[];
/** {@inheritDoc SavedObjectsMigrationVersion} */
migrationVersion?: SavedObjectsMigrationVersion;
Expand Down
3 changes: 3 additions & 0 deletions src/core/server/server.api.md
Original file line number Diff line number Diff line change
Expand Up @@ -1945,6 +1945,8 @@ export interface SavedObjectsBulkCreateObject<T = unknown> {
references?: SavedObjectReference[];
// (undocumented)
type: string;
// (undocumented)
version?: string;
}

// @public (undocumented)
Expand Down Expand Up @@ -2079,6 +2081,7 @@ export interface SavedObjectsCreateOptions extends SavedObjectsBaseOptions {
// (undocumented)
references?: SavedObjectReference[];
refresh?: MutatingOperationRefreshSetting;
version?: string;
}

// @public (undocumented)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,7 @@ export function createInstallRoute(
let createResults;
try {
createResults = await context.core.savedObjects.client.bulkCreate(
sampleDataset.savedObjects,
sampleDataset.savedObjects.map(({ version, ...savedObject }) => savedObject),
{ overwrite: true }
);
} catch (err) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,9 @@ import {
import { deleteKibanaSavedObjectsAssets } from '../../packages/remove';
import { getInstallationObject, savedObjectTypes } from '../../packages';

type SavedObjectToBe = Required<SavedObjectsBulkCreateObject> & { type: AssetType };
type SavedObjectToBe = Required<Omit<SavedObjectsBulkCreateObject, 'version'>> & {
type: AssetType;
};
export type ArchiveAsset = Pick<
SavedObject,
'id' | 'attributes' | 'migrationVersion' | 'references'
Expand Down

0 comments on commit 00836e5

Please sign in to comment.