Initial implementation of a generic ndjson collator.
Signed-off-by: Eric Peterson <ericpeterson@spotify.com>
This commit is contained in:
Eric Peterson
+157
@@ -0,0 +1,157 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import {
|
||||
getVoidLogger,
|
||||
ReadUrlResponse,
|
||||
UrlReader,
|
||||
UrlReaders,
|
||||
} from '@backstage/backend-common';
|
||||
import { ConfigReader } from '@backstage/config';
|
||||
import { Readable } from 'stream';
|
||||
import { NewlineDelimitedJsonCollatorFactory } from './NewlineDelimitedJsonCollatorFactory';
|
||||
import { TestPipeline } from '../test-utils';
|
||||
|
||||
describe('DefaultCatalogCollatorFactory', () => {
|
||||
const config = new ConfigReader({});
|
||||
const logger = getVoidLogger();
|
||||
|
||||
it('has expected type', () => {
|
||||
const factory = NewlineDelimitedJsonCollatorFactory.fromConfig(config, {
|
||||
type: 'expected-type',
|
||||
searchPattern: 'test://folder/prefix-*',
|
||||
logger,
|
||||
reader: UrlReaders.default({ logger, config }),
|
||||
});
|
||||
expect(factory.type).toBe('expected-type');
|
||||
});
|
||||
|
||||
describe('getCollator', () => {
|
||||
let readable: Readable;
|
||||
let reader: jest.Mocked<
|
||||
UrlReader & { readUrl: jest.Mock<Promise<ReadUrlResponse>> }
|
||||
>;
|
||||
let factory: NewlineDelimitedJsonCollatorFactory;
|
||||
|
||||
beforeEach(async () => {
|
||||
jest.clearAllMocks();
|
||||
|
||||
readable = new Readable();
|
||||
readable._read = () => {};
|
||||
reader = {
|
||||
search: jest.fn(),
|
||||
read: jest.fn(),
|
||||
readTree: jest.fn(),
|
||||
readUrl: jest.fn(),
|
||||
};
|
||||
factory = NewlineDelimitedJsonCollatorFactory.fromConfig(config, {
|
||||
type: 'expected-type',
|
||||
searchPattern: 'test://folder/prefix-*',
|
||||
logger,
|
||||
reader: UrlReaders.create({
|
||||
logger,
|
||||
config,
|
||||
factories: [() => [{ predicate: () => true, reader }]],
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
it('throws if url reader throws an error during search', async () => {
|
||||
reader.search.mockRejectedValue(new Error('Expected error'));
|
||||
|
||||
await expect(() => factory.getCollator()).rejects.toThrowError(
|
||||
'Expected error',
|
||||
);
|
||||
});
|
||||
|
||||
it('throws if no matching files are found', async () => {
|
||||
reader.search.mockResolvedValue({ files: [], etag: '' });
|
||||
|
||||
await expect(() => factory.getCollator()).rejects.toThrowError(
|
||||
'Could not find an .ndjson file matching',
|
||||
);
|
||||
});
|
||||
|
||||
it('throws if matching file is not .ndjson', async () => {
|
||||
reader.search.mockResolvedValue({
|
||||
files: [{ url: 'test://folder/prefix-1.avro', content: jest.fn() }],
|
||||
etag: '',
|
||||
});
|
||||
reader.readUrl.mockResolvedValue({
|
||||
buffer: jest.fn(),
|
||||
stream: jest.fn().mockReturnValue(readable),
|
||||
});
|
||||
|
||||
await expect(() => factory.getCollator()).rejects.toThrowError(
|
||||
'Could not find an .ndjson file matching',
|
||||
);
|
||||
});
|
||||
|
||||
it('gets stream using latest matched url', async () => {
|
||||
reader.search.mockResolvedValue({
|
||||
files: [
|
||||
{ url: 'test://folder/prefix-1.ndjson', content: jest.fn() },
|
||||
{ url: 'test://folder/prefix-2.ndjson', content: jest.fn() },
|
||||
],
|
||||
etag: '',
|
||||
});
|
||||
reader.readUrl.mockResolvedValue({
|
||||
buffer: jest.fn(),
|
||||
stream: jest.fn().mockReturnValue(readable),
|
||||
});
|
||||
|
||||
await factory.getCollator();
|
||||
|
||||
expect(reader.search).toHaveBeenCalledWith(
|
||||
'test://folder/prefix-*',
|
||||
undefined,
|
||||
);
|
||||
expect(reader.readUrl).toHaveBeenCalledWith(
|
||||
'test://folder/prefix-2.ndjson',
|
||||
undefined,
|
||||
);
|
||||
});
|
||||
|
||||
it('transforms newline delimited json into readable stream of documents', async () => {
|
||||
reader.search.mockResolvedValue({
|
||||
files: [{ url: 'test://folder/prefix-1.ndjson', content: jest.fn() }],
|
||||
etag: '',
|
||||
});
|
||||
reader.readUrl.mockResolvedValue({
|
||||
buffer: jest.fn(),
|
||||
stream: jest
|
||||
.fn()
|
||||
.mockReturnValue(
|
||||
Readable.from(
|
||||
'{"title": "Title 1", "location": "/title-1", "text": "text 1"}\n{"title": "Title 2", "location": "/title-2", "text": "text 2"}',
|
||||
),
|
||||
),
|
||||
});
|
||||
|
||||
const collator = await factory.getCollator();
|
||||
const pipeline = TestPipeline.withSubject(collator);
|
||||
const { documents } = await pipeline.execute();
|
||||
|
||||
expect(documents).toHaveLength(2);
|
||||
expect(documents[0].title).toBe('Title 1');
|
||||
expect(documents[0].location).toBe('/title-1');
|
||||
expect(documents[0].text).toBe('text 1');
|
||||
expect(documents[1].title).toBe('Title 2');
|
||||
expect(documents[1].location).toBe('/title-2');
|
||||
expect(documents[1].text).toBe('text 2');
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,147 @@
|
||||
/*
|
||||
* Copyright 2021 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import { UrlReader } from '@backstage/backend-common';
|
||||
import { Config } from '@backstage/config';
|
||||
import { Permission } from '@backstage/plugin-permission-common';
|
||||
import { DocumentCollatorFactory } from '@backstage/plugin-search-common';
|
||||
import { parse as parseNdjson } from 'ndjson';
|
||||
import { Readable } from 'stream';
|
||||
import { Logger } from 'winston';
|
||||
|
||||
/**
|
||||
* @beta
|
||||
*/
|
||||
export type NewlineDelimitedJsonCollatorFactoryOptions = {
|
||||
type: string;
|
||||
searchPattern: string;
|
||||
reader: UrlReader;
|
||||
logger: Logger;
|
||||
visibilityPermission?: Permission;
|
||||
};
|
||||
|
||||
/**
|
||||
* Factory class producing a collator that can be used to index documents
|
||||
* sourced from the latest newline delimited JSON file matching a given search
|
||||
* pattern. "Latest" is determined by the name of the file (last alphabetically
|
||||
* is considered latest).
|
||||
*
|
||||
* @remarks
|
||||
* The reader provided must implement the `search()` method as well as the
|
||||
* `readUrl` method whose response includes the `stream()` method. Naturally,
|
||||
* the reader must also be configured to understand the given search pattern.
|
||||
*
|
||||
* @example
|
||||
* Here's an example configuration using Google Cloud Storage, which would
|
||||
* return the latest file under the `bucket` GCS bucket with files like
|
||||
* `xyz-2021.ndjson` or `xyz-2022.ndjson`.
|
||||
* ```ts
|
||||
* indexBuilder.addCollator({
|
||||
* schedule,
|
||||
* factory: NewlineDelimitedJsonCollatorFactory.fromConfig(env.config, {
|
||||
* type: 'techdocs',
|
||||
* searchPattern: 'https://storage.cloud.google.com/bucket/xyz-*',
|
||||
* reader: env.reader,
|
||||
* logger: env.logger,
|
||||
* })
|
||||
* });
|
||||
* ```
|
||||
*
|
||||
* @beta
|
||||
*/
|
||||
export class NewlineDelimitedJsonCollatorFactory
|
||||
implements DocumentCollatorFactory
|
||||
{
|
||||
/** {@inheritDoc @backstage/plugin-search-common#DocumentCollatorFactory."type"} */
|
||||
readonly type: string;
|
||||
|
||||
/** {@inheritDoc @backstage/plugin-search-common#DocumentCollatorFactory.visibilityPermission} */
|
||||
public readonly visibilityPermission: Permission | undefined;
|
||||
|
||||
private constructor(
|
||||
type: string,
|
||||
private readonly searchPattern: string,
|
||||
private readonly reader: UrlReader,
|
||||
private readonly logger: Logger,
|
||||
visibilityPermission: Permission | undefined,
|
||||
) {
|
||||
this.type = type;
|
||||
this.visibilityPermission = visibilityPermission;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a NewlineDelimitedJsonCollatorFactory instance from configuration
|
||||
* and a set of options.
|
||||
*/
|
||||
static fromConfig(
|
||||
_config: Config,
|
||||
options: NewlineDelimitedJsonCollatorFactoryOptions,
|
||||
): NewlineDelimitedJsonCollatorFactory {
|
||||
return new NewlineDelimitedJsonCollatorFactory(
|
||||
options.type,
|
||||
options.searchPattern,
|
||||
options.reader,
|
||||
options.logger,
|
||||
options.visibilityPermission,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the "latest" URL for the given search pattern (e.g. the one at the
|
||||
* end of the list, sorted alphabetically).
|
||||
*/
|
||||
private async lastUrl(): Promise<string | undefined> {
|
||||
try {
|
||||
// Search for files matching the given pattern, then sort/reverse. The
|
||||
// first item in the list will be the "latest" file.
|
||||
this.logger.info(
|
||||
`Attempting to find latest .ndjson matching ${this.searchPattern}`,
|
||||
);
|
||||
const { files } = await this.reader.search(this.searchPattern);
|
||||
const candidates = files
|
||||
.filter(file => file.url.endsWith('.ndjson'))
|
||||
.sort((a, b) => a.url.localeCompare(b.url))
|
||||
.reverse();
|
||||
|
||||
return candidates[0]?.url;
|
||||
} catch (e) {
|
||||
this.logger.error(`Could not search for ${this.searchPattern}`, e);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
/** {@inheritDoc @backstage/plugin-search-common#DocumentCollatorFactory.getCollator} */
|
||||
async getCollator(): Promise<Readable> {
|
||||
// Search for files matching the given pattern.
|
||||
const lastUrl = await this.lastUrl();
|
||||
|
||||
// Abort if no such file could be found.
|
||||
if (!lastUrl) {
|
||||
const noMatchingFile = `Could not find an .ndjson file matching ${this.searchPattern}`;
|
||||
this.logger.error(noMatchingFile);
|
||||
throw new Error(noMatchingFile);
|
||||
} else {
|
||||
this.logger.info(`Using latest .ndjson file ${lastUrl}`);
|
||||
}
|
||||
|
||||
// Use the UrlReader to try and stream the file.
|
||||
const readerResponse = await this.reader.readUrl!(lastUrl);
|
||||
const stream = readerResponse.stream!();
|
||||
|
||||
// Use ndjson's parser to turn the raw file into an object-mode stream.
|
||||
return stream.pipe(parseNdjson());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
/*
|
||||
* Copyright 2021 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
export type { NewlineDelimitedJsonCollatorFactoryOptions } from './NewlineDelimitedJsonCollatorFactory';
|
||||
|
||||
export { NewlineDelimitedJsonCollatorFactory } from './NewlineDelimitedJsonCollatorFactory';
|
||||
@@ -22,6 +22,7 @@
|
||||
|
||||
export { IndexBuilder } from './IndexBuilder';
|
||||
export { Scheduler } from './Scheduler';
|
||||
export * from './collators';
|
||||
export { LunrSearchEngine } from './engines';
|
||||
export type {
|
||||
ConcreteLunrQuery,
|
||||
|
||||
@@ -66,14 +66,14 @@ export class TestPipeline {
|
||||
return new TestPipeline({ decorator: subject });
|
||||
}
|
||||
|
||||
if (subject instanceof Readable) {
|
||||
return new TestPipeline({ collator: subject });
|
||||
}
|
||||
|
||||
if (subject instanceof Writable) {
|
||||
return new TestPipeline({ indexer: subject });
|
||||
}
|
||||
|
||||
if (subject.readable || subject instanceof Readable) {
|
||||
return new TestPipeline({ collator: subject });
|
||||
}
|
||||
|
||||
throw new Error(
|
||||
'Unknown test subject: are you passing a readable, writable, or transform stream?',
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user