[EPM] Implement getConfig for dataset (#53261)

* [EPM] Implement getConfig for dataset

* Implements a getConfig method on a dataset object. 
* Build the configuration for each dataset in a package.
* construct and save streams into datasource saved object

x-pack/legacy/plugins/epm/common/types.ts

@@ -16,7 +16,7 @@ export enum InstallationStatus {
 }
 
 export type ServiceName = 'kibana' | 'elasticsearch';
-export type AssetType = KibanaAssetType | ElasticsearchAssetType;
+export type AssetType = KibanaAssetType | ElasticsearchAssetType | AgentAssetType;
 
 export enum KibanaAssetType {
   dashboard = 'dashboard',
@@ -31,6 +31,10 @@ export enum ElasticsearchAssetType {
   ilmPolicy = 'ilm-policy',
 }
 
+export enum AgentAssetType {
+  input = 'input',
+}
+
 // from /package/{name}
 // type Package struct at https://github.com/elastic/package-registry/blob/master/util/package.go
 // https://github.com/elastic/package-registry/blob/master/docs/api/package.json
@@ -120,12 +124,17 @@ export interface Dataset {
   name: string;
   release: string;
   ingeset_pipeline: string;
-  vars: object[];
+  vars: VarsEntry[];
   type: string;
   // This is for convenience and not in the output from the registry. When creating a dataset, this info should be added.
   package: string;
 }
 
+export interface VarsEntry {
+  name: string;
+  default: string;
+}
+
 // some properties are optional in Registry responses but required in EPM
 // internal until we need them
 interface PackageAdditions {

x-pack/legacy/plugins/epm/public/constants.ts

@@ -23,6 +23,7 @@ export const AssetTitleMap: Record<AssetType, string> = {
   'index-template': 'Index Template',
   search: 'Saved Search',
   visualization: 'Visualization',
+  input: 'Agent input',
 };
 
 export const ServiceTitleMap: Record<ServiceName, string> = {

x-pack/legacy/plugins/epm/server/datasources/create.ts

@@ -5,8 +5,9 @@
  */
 
 import fetch from 'node-fetch';
+import yaml from 'js-yaml';
 import { SavedObjectsClientContract } from 'src/core/server/';
-import { Asset, Datasource, InputType } from '../../../ingest/server/libs/types';
+import { Asset, Datasource, Stream } from '../../../ingest/server/libs/types';
 import { SAVED_OBJECT_TYPE_DATASOURCES } from '../../common/constants';
 import { AssetReference, Dataset, InstallationStatus, RegistryPackage } from '../../common/types';
 import { CallESAsCurrentUser } from '../lib/cluster_access';
@@ -16,6 +17,7 @@ import { installTemplates } from '../lib/elasticsearch/template/install';
 import { getPackageInfo, PackageNotInstalledError } from '../packages';
 import * as Registry from '../registry';
 import { Request } from '../types';
+import { createInput } from '../lib/agent/agent';
 
 export async function createDatasource(options: {
   savedObjectsClient: SavedObjectsClientContract;
@@ -38,6 +40,8 @@ export async function createDatasource(options: {
   await baseSetup(callCluster);
   const pkg = await Registry.fetchInfo(pkgkey);
 
+  const streams = await getStreams(pkgkey, datasets);
+
   await Promise.all([
     installTemplates(pkg, callCluster),
     saveDatasourceReferences({
@@ -47,6 +51,7 @@ export async function createDatasource(options: {
       datasets,
       toSave,
       request,
+      streams,
     }),
   ]);
 
@@ -73,8 +78,9 @@ async function saveDatasourceReferences(options: {
   datasourceName: string;
   toSave: AssetReference[];
   request: Request;
+  streams: Stream[];
 }) {
-  const { savedObjectsClient, pkg, toSave, datasets, datasourceName, request } = options;
+  const { savedObjectsClient, pkg, toSave, datasets, datasourceName, request, streams } = options;
   const savedDatasource = await getDatasource({ savedObjectsClient, pkg });
   const savedAssets = savedDatasource?.package.assets;
   const assetsReducer = (current: Asset[] = [], pending: Asset) => {
@@ -89,7 +95,9 @@ async function saveDatasourceReferences(options: {
     datasourceName,
     datasets,
     assets: toInstall,
+    streams,
   });
+
   // ideally we'd call .create from /x-pack/legacy/plugins/ingest/server/libs/datasources.ts#L22
   // or something similar, but it's a class not an object so many pieces are missing
   // we'd still need `user` from the request object, but that's not terrible
@@ -99,6 +107,23 @@ async function saveDatasourceReferences(options: {
   return toInstall;
 }
 
+async function getStreams(pkgkey: string, datasets: Dataset[]) {
+  const streams: Stream[] = [];
+  if (datasets) {
+    for (const dataset of datasets) {
+      const input = yaml.load(await getConfig(pkgkey, dataset));
+      if (input) {
+        streams.push({
+          id: dataset.name,
+          input,
+          output_id: 'default',
+        });
+      }
+    }
+  }
+  return streams;
+}
+
 async function getDatasource(options: {
   savedObjectsClient: SavedObjectsClientContract;
   pkg: RegistryPackage;
@@ -116,29 +141,16 @@ interface CreateFakeDatasource {
   datasourceName: string;
   datasets: Dataset[];
   assets: Asset[] | undefined;
+  streams: Stream[];
 }
 
 function createFakeDatasource({
   pkg,
   datasourceName,
   datasets,
   assets = [],
+  streams,
 }: CreateFakeDatasource): Datasource {
-  const streams = datasets.map(dataset => ({
-    id: dataset.name,
-    input: {
-      type: InputType.Log,
-      config: { config: 'values', go: 'here' },
-      ingest_pipelines: ['string'],
-      id: 'string',
-      index_template: 'string',
-      ilm_policy: 'string',
-      fields: [{}],
-    },
-    config: { config: 'values', go: 'here' },
-    output_id: 'output_id',
-    processors: ['string'],
-  }));
   return {
     id: Registry.pkgToPkgKey(pkg),
     name: datasourceName,
@@ -173,7 +185,7 @@ async function ingestDatasourceCreate({
   const url = `${origin}${basePath}${apiPath}`;
   const body = { datasource };
   delete request.headers['transfer-encoding'];
-  return fetch(url, {
+  await fetch(url, {
     method: 'post',
     body: JSON.stringify(body),
     headers: {
@@ -182,5 +194,28 @@ async function ingestDatasourceCreate({
       // the main (only?) one we want is `authorization`
       ...request.headers,
     },
-  }).then(response => response.json());
+  });
 }
+
+async function getConfig(pkgkey: string, dataset: Dataset): Promise<string> {
+  const vars = dataset.vars;
+
+  // This searches for the /agent/input.yml file
+  const paths = await Registry.getArchiveInfo(pkgkey, (entry: Registry.ArchiveEntry) =>
+    isDatasetInput(entry, dataset.name)
+  );
+
+  if (paths.length === 1) {
+    const buffer = Registry.getAsset(paths[0]);
+    // Load input template from path
+    return createInput(vars, buffer.toString());
+  }
+  return '';
+}
+
+const isDatasetInput = ({ path }: Registry.ArchiveEntry, datasetName: string) => {
+  const pathParts = Registry.pathParts(path);
+  return !isDirectory({ path }) && pathParts.type === 'input' && pathParts.dataset === datasetName;
+};
+
+const isDirectory = ({ path }: Registry.ArchiveEntry) => path.endsWith('/');

x-pack/legacy/plugins/epm/server/lib/agent/agent.test.ts

@@ -15,7 +15,7 @@ test('test converting input and manifest into template', () => {
   );
 
   const inputTemplate = fs.readFileSync(path.join(__dirname, 'tests/input.yml'), 'utf8');
-  const output = createInput(manifest, inputTemplate);
+  const output = createInput(manifest.vars, inputTemplate);
 
   // Golden file path
   const generatedFile = path.join(__dirname, './tests/input.generated.yaml');

x-pack/legacy/plugins/epm/server/lib/agent/agent.ts

@@ -5,23 +5,16 @@
  */
 
 import Handlebars from 'handlebars';
+import { VarsEntry } from '../../../common/types';
 
-interface Manifest {
-  vars: VarsEntry[];
-}
-
-interface VarsEntry {
-  name: string;
-  default: string;
-}
 /**
- * This takes a manifest object as input and merges it with the input template.
+ * This takes a dataset object as input and merges it with the input template.
  * It returns the resolved template as a string.
  */
-export function createInput(manifest: Manifest, inputTemplate: string): string {
+export function createInput(vars: VarsEntry[], inputTemplate: string): string {
   const view: Record<VarsEntry['name'], VarsEntry['default']> = {};
 
-  for (const v of manifest.vars) {
+  for (const v of vars) {
     view[v.name] = v.default;
   }
 

x-pack/legacy/plugins/fleet/server/libs/policy.ts

@@ -16,7 +16,7 @@ export class PolicyLib {
     return flatten(
       datasources.map((ds: Datasource) => {
         return ds.streams.map(stream => ({
-          ...stream.config,
+          ...stream.input,
           id: stream.id,
           type: stream.input.type as any,
           output: { use_output: stream.output_id },