Skip to content

Commit

Permalink
Add support for input[].dataset.type and change stream.dataset ma…
Browse files Browse the repository at this point in the history
…pping to be object containing `name` with instead of just a string
  • Loading branch information
jen-huang committed Jun 24, 2020
1 parent fe5a6ae commit f34e642
Show file tree
Hide file tree
Showing 15 changed files with 105 additions and 50 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ describe('Ingest Manager - storedDatasourcesToAgentInputs', () => {
{
id: 'test-logs-foo',
enabled: true,
dataset: 'foo',
dataset: { name: 'foo' },
vars: {
fooVar: { value: 'foo-value' },
fooVar2: { value: [1, 2] },
Expand All @@ -52,7 +52,7 @@ describe('Ingest Manager - storedDatasourcesToAgentInputs', () => {
{
id: 'test-logs-bar',
enabled: true,
dataset: 'bar',
dataset: { name: 'bar' },
vars: {
barVar: { value: 'bar-value' },
barVar2: { value: [1, 2] },
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,10 @@ export const storedDatasourcesToAgentInputs = (
id: datasource.id || datasource.name,
name: datasource.name,
type: input.type,
dataset: { namespace: datasource.namespace || 'default' },
dataset: {
namespace: datasource.namespace || 'default',
...(input.dataset?.type ? { type: input.dataset.type } : {}),
},
use_output: DEFAULT_OUTPUT.name,
...Object.entries(input.config || {}).reduce((acc, [key, { value }]) => {
acc[key] = value;
Expand All @@ -35,7 +38,7 @@ export const storedDatasourcesToAgentInputs = (
.map((stream) => {
const fullStream: FullAgentConfigInputStream = {
id: stream.id,
dataset: { name: stream.dataset },
dataset: stream.dataset,
...stream.agent_stream,
...Object.entries(stream.config || {}).reduce((acc, [key, { value }]) => {
acc[key] = value;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -82,15 +82,17 @@ describe('Ingest Manager - packageToConfig', () => {
).toEqual([
{
type: 'foo',
dataset: { type: 'logs' },
enabled: true,
streams: [{ id: 'foo-foo', enabled: true, dataset: 'foo' }],
streams: [{ id: 'foo-foo', enabled: true, dataset: { name: 'foo' } }],
},
{
type: 'bar',
dataset: { type: 'logs' },
enabled: true,
streams: [
{ id: 'bar-bar', enabled: true, dataset: 'bar' },
{ id: 'bar-bar2', enabled: true, dataset: 'bar2' },
{ id: 'bar-bar', enabled: true, dataset: { name: 'bar' } },
{ id: 'bar-bar2', enabled: true, dataset: { name: 'bar2' } },
],
},
]);
Expand Down Expand Up @@ -136,30 +138,32 @@ describe('Ingest Manager - packageToConfig', () => {
).toEqual([
{
type: 'foo',
dataset: { type: 'logs' },
enabled: true,
streams: [
{
id: 'foo-foo',
enabled: true,
dataset: 'foo',
dataset: { name: 'foo' },
vars: { 'var-name': { value: 'foo-var-value' } },
},
],
},
{
type: 'bar',
dataset: { type: 'logs' },
enabled: true,
streams: [
{
id: 'bar-bar',
enabled: true,
dataset: 'bar',
dataset: { name: 'bar' },
vars: { 'var-name': { type: 'text', value: 'bar-var-value' } },
},
{
id: 'bar-bar2',
enabled: true,
dataset: 'bar2',
dataset: { name: 'bar2' },
vars: { 'var-name': { type: 'yaml', value: 'bar2-var-value' } },
},
],
Expand Down Expand Up @@ -247,6 +251,7 @@ describe('Ingest Manager - packageToConfig', () => {
).toEqual([
{
type: 'foo',
dataset: { type: 'logs' },
enabled: true,
vars: {
'foo-input-var-name': { value: 'foo-input-var-value' },
Expand All @@ -257,7 +262,7 @@ describe('Ingest Manager - packageToConfig', () => {
{
id: 'foo-foo',
enabled: true,
dataset: 'foo',
dataset: { name: 'foo' },
vars: {
'var-name': { value: 'foo-var-value' },
},
Expand All @@ -266,6 +271,7 @@ describe('Ingest Manager - packageToConfig', () => {
},
{
type: 'bar',
dataset: { type: 'logs' },
enabled: true,
vars: {
'bar-input-var-name': { value: ['value1', 'value2'] },
Expand All @@ -275,15 +281,15 @@ describe('Ingest Manager - packageToConfig', () => {
{
id: 'bar-bar',
enabled: true,
dataset: 'bar',
dataset: { name: 'bar' },
vars: {
'var-name': { value: 'bar-var-value' },
},
},
{
id: 'bar-bar2',
enabled: true,
dataset: 'bar2',
dataset: { name: 'bar2' },
vars: {
'var-name': { value: 'bar2-var-value' },
},
Expand All @@ -292,20 +298,21 @@ describe('Ingest Manager - packageToConfig', () => {
},
{
type: 'with-disabled-streams',
dataset: { type: 'logs' },
enabled: false,
streams: [
{
id: 'with-disabled-streams-disabled',
enabled: false,
dataset: 'disabled',
dataset: { name: 'disabled' },
vars: {
'var-name': { value: [] },
},
},
{
id: 'with-disabled-streams-disabled2',
enabled: false,
dataset: 'disabled2',
dataset: { name: 'disabled2' },
},
],
},
Expand Down
23 changes: 18 additions & 5 deletions x-pack/plugins/ingest_manager/common/services/package_to_config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
* or more contributor license agreements. Licensed under the Elastic License;
* you may not use this file except in compliance with the Elastic License.
*/
import { pluck, uniq } from 'lodash';
import {
PackageInfo,
RegistryConfigTemplate,
Expand All @@ -16,18 +17,19 @@ import {
NewDatasource,
} from '../types';

const findStreamsForInputType = (
const getStreamsForInputType = (
inputType: string,
packageInfo: PackageInfo
): Array<RegistryStream & { dataset: { name: string } }> => {
const streams: Array<RegistryStream & { dataset: { name: string } }> = [];
): Array<RegistryStream & { dataset: { type: string; name: string } }> => {
const streams: Array<RegistryStream & { dataset: { type: string; name: string } }> = [];

(packageInfo.datasets || []).forEach((dataset) => {
(dataset.streams || []).forEach((stream) => {
if (stream.input === inputType) {
streams.push({
...stream,
dataset: {
type: dataset.type,
name: dataset.name,
},
});
Expand All @@ -38,6 +40,11 @@ const findStreamsForInputType = (
return streams;
};

const getDatasetTypeForInputType = (inputType: string, packageInfo: PackageInfo): string => {
const datasetTypes = uniq(pluck(getStreamsForInputType(inputType, packageInfo), 'dataset.type'));
return datasetTypes[0];
};

/*
* This service creates a datasource inputs definition from defaults provided in package info
*/
Expand Down Expand Up @@ -70,14 +77,14 @@ export const packageToConfigDatasourceInputs = (packageInfo: PackageInfo): Datas
};

// Map each package input stream into datasource input stream
const streams: DatasourceInputStream[] = findStreamsForInputType(
const streams: DatasourceInputStream[] = getStreamsForInputType(
packageInput.type,
packageInfo
).map((packageStream) => {
const stream: DatasourceInputStream = {
id: `${packageInput.type}-${packageStream.dataset.name}`,
enabled: packageStream.enabled === false ? false : true,
dataset: packageStream.dataset.name,
dataset: { name: packageStream.dataset.name },
};
if (packageStream.vars && packageStream.vars.length) {
stream.vars = packageStream.vars.reduce(varsReducer, {});
Expand All @@ -91,6 +98,12 @@ export const packageToConfigDatasourceInputs = (packageInfo: PackageInfo): Datas
streams,
};

const datasetType = getDatasetTypeForInputType(packageInput.type, packageInfo);

if (datasetType) {
input.dataset = { type: datasetType };
}

if (packageInput.vars && packageInput.vars.length) {
input.vars = packageInput.vars.reduce(varsReducer, {});
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,9 @@ export type DatasourceConfigRecord = Record<string, DatasourceConfigRecordEntry>
export interface NewDatasourceInputStream {
id: string;
enabled: boolean;
dataset: string;
dataset: {
name: string;
};
processors?: string[];
config?: DatasourceConfigRecord;
vars?: DatasourceConfigRecord;
Expand All @@ -32,6 +34,9 @@ export interface DatasourceInputStream extends NewDatasourceInputStream {

export interface NewDatasourceInput {
type: string;
dataset?: {
type: string;
};
enabled: boolean;
processors?: string[];
config?: DatasourceConfigRecord;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ export const DatasourceInputPanel: React.FunctionComponent<{
<EuiFlexGroup direction="column">
{packageInputStreams.map((packageInputStream) => {
const datasourceInputStream = datasourceInput.streams.find(
(stream) => stream.dataset === packageInputStream.dataset.name
(stream) => stream.dataset.name === packageInputStream.dataset.name
);
return datasourceInputStream ? (
<EuiFlexItem key={packageInputStream.dataset.name}>
Expand All @@ -186,7 +186,7 @@ export const DatasourceInputPanel: React.FunctionComponent<{
datasourceInputStream={datasourceInputStream}
updateDatasourceInputStream={(updatedStream: Partial<DatasourceInputStream>) => {
const indexOfUpdatedStream = datasourceInput.streams.findIndex(
(stream) => stream.dataset === packageInputStream.dataset.name
(stream) => stream.dataset.name === packageInputStream.dataset.name
);
const newStreams = [...datasourceInput.streams];
newStreams[indexOfUpdatedStream] = {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ describe('Ingest Manager - validateDatasource()', () => {
streams: [
{
id: 'foo-foo',
dataset: 'foo',
dataset: { name: 'foo' },
enabled: true,
vars: { 'var-name': { value: 'test_yaml: value', type: 'yaml' } },
},
Expand All @@ -174,13 +174,13 @@ describe('Ingest Manager - validateDatasource()', () => {
streams: [
{
id: 'bar-bar',
dataset: 'bar',
dataset: { name: 'bar' },
enabled: true,
vars: { 'var-name': { value: 'test_yaml: value', type: 'yaml' } },
},
{
id: 'bar-bar2',
dataset: 'bar2',
dataset: { name: 'bar2' },
enabled: true,
vars: { 'var-name': { value: undefined, type: 'text' } },
},
Expand All @@ -197,13 +197,13 @@ describe('Ingest Manager - validateDatasource()', () => {
streams: [
{
id: 'with-disabled-streams-disabled',
dataset: 'disabled',
dataset: { name: 'disabled' },
enabled: false,
vars: { 'var-name': { value: undefined, type: 'text' } },
},
{
id: 'with-disabled-streams-disabled-without-vars',
dataset: 'disabled2',
dataset: { name: 'disabled2' },
enabled: false,
},
],
Expand All @@ -217,7 +217,7 @@ describe('Ingest Manager - validateDatasource()', () => {
streams: [
{
id: 'with-no-stream-vars-bar',
dataset: 'bar',
dataset: { name: 'bar' },
enabled: true,
},
],
Expand All @@ -240,7 +240,7 @@ describe('Ingest Manager - validateDatasource()', () => {
streams: [
{
id: 'foo-foo',
dataset: 'foo',
dataset: { name: 'foo' },
enabled: true,
vars: { 'var-name': { value: 'invalidyaml: test\n foo bar:', type: 'yaml' } },
},
Expand All @@ -256,13 +256,13 @@ describe('Ingest Manager - validateDatasource()', () => {
streams: [
{
id: 'bar-bar',
dataset: 'bar',
dataset: { name: 'bar' },
enabled: true,
vars: { 'var-name': { value: ' \n\n', type: 'yaml' } },
},
{
id: 'bar-bar2',
dataset: 'bar2',
dataset: { name: 'bar2' },
enabled: true,
vars: { 'var-name': { value: undefined, type: 'text' } },
},
Expand All @@ -279,7 +279,7 @@ describe('Ingest Manager - validateDatasource()', () => {
streams: [
{
id: 'with-disabled-streams-disabled',
dataset: 'disabled',
dataset: { name: 'disabled' },
enabled: false,
vars: {
'var-name': {
Expand All @@ -290,7 +290,7 @@ describe('Ingest Manager - validateDatasource()', () => {
},
{
id: 'with-disabled-streams-disabled-without-vars',
dataset: 'disabled2',
dataset: { name: 'disabled2' },
enabled: false,
},
],
Expand All @@ -304,7 +304,7 @@ describe('Ingest Manager - validateDatasource()', () => {
streams: [
{
id: 'with-no-stream-vars-bar',
dataset: 'bar',
dataset: { name: 'bar' },
enabled: true,
},
],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,7 @@ export const validateDatasource = (
if (stream.vars) {
const streamVarsByName = (
(
registryStreamsByDataset[stream.dataset].find(
registryStreamsByDataset[stream.dataset.name].find(
(registryStream) => registryStream.input === input.type
) || {}
).vars || []
Expand Down
Loading

0 comments on commit f34e642

Please sign in to comment.