diff --git a/x-pack/plugins/ingest_manager/common/services/config_to_yaml.ts b/x-pack/plugins/ingest_manager/common/services/config_to_yaml.ts
index a3bef72e8db5a3..c2043a40369e28 100644
--- a/x-pack/plugins/ingest_manager/common/services/config_to_yaml.ts
+++ b/x-pack/plugins/ingest_manager/common/services/config_to_yaml.ts
@@ -16,7 +16,7 @@ const CONFIG_KEYS_ORDER = [
   'inputs',
   'enabled',
   'use_output',
-  'package',
+  'meta',
   'input',
 ];
 
diff --git a/x-pack/plugins/ingest_manager/common/services/datasources_to_agent_inputs.test.ts b/x-pack/plugins/ingest_manager/common/services/datasources_to_agent_inputs.test.ts
index df94168ec88d02..538951ff103992 100644
--- a/x-pack/plugins/ingest_manager/common/services/datasources_to_agent_inputs.test.ts
+++ b/x-pack/plugins/ingest_manager/common/services/datasources_to_agent_inputs.test.ts
@@ -39,7 +39,7 @@ describe('Ingest Manager - storedDatasourcesToAgentInputs', () => {
       {
         id: 'test-logs-foo',
         enabled: true,
-        dataset: 'foo',
+        dataset: { name: 'foo', type: 'logs' },
         vars: {
           fooVar: { value: 'foo-value' },
           fooVar2: { value: [1, 2] },
@@ -52,7 +52,7 @@ describe('Ingest Manager - storedDatasourcesToAgentInputs', () => {
       {
         id: 'test-logs-bar',
         enabled: true,
-        dataset: 'bar',
+        dataset: { name: 'bar', type: 'logs' },
         vars: {
           barVar: { value: 'bar-value' },
           barVar2: { value: [1, 2] },
@@ -101,23 +101,41 @@ describe('Ingest Manager - storedDatasourcesToAgentInputs', () => {
   });
 
   it('returns agent inputs', () => {
-    expect(storedDatasourcesToAgentInputs([{ ...mockDatasource, inputs: [mockInput] }])).toEqual([
+    expect(
+      storedDatasourcesToAgentInputs([
+        {
+          ...mockDatasource,
+          package: {
+            name: 'mock-package',
+            title: 'Mock package',
+            version: '0.0.0',
+          },
+          inputs: [mockInput],
+        },
+      ])
+    ).toEqual([
       {
         id: 'some-uuid',
         name: 'mock-datasource',
         type: 'test-logs',
         dataset: { namespace: 'default' },
         use_output: 'default',
+        meta: {
+          package: {
+            name: 'mock-package',
+            version: '0.0.0',
+          },
+        },
         streams: [
           {
             id: 'test-logs-foo',
-            dataset: { name: 'foo' },
+            dataset: { name: 'foo', type: 'logs' },
             fooKey: 'fooValue1',
             fooKey2: ['fooValue2'],
           },
           {
             id: 'test-logs-bar',
-            dataset: { name: 'bar' },
+            dataset: { name: 'bar', type: 'logs' },
           },
         ],
       },
@@ -147,7 +165,7 @@ describe('Ingest Manager - storedDatasourcesToAgentInputs', () => {
         streams: [
           {
             id: 'test-logs-foo',
-            dataset: { name: 'foo' },
+            dataset: { name: 'foo', type: 'logs' },
             fooKey: 'fooValue1',
             fooKey2: ['fooValue2'],
           },
diff --git a/x-pack/plugins/ingest_manager/common/services/datasources_to_agent_inputs.ts b/x-pack/plugins/ingest_manager/common/services/datasources_to_agent_inputs.ts
index d5a752e817b4f2..c6c5d784396db3 100644
--- a/x-pack/plugins/ingest_manager/common/services/datasources_to_agent_inputs.ts
+++ b/x-pack/plugins/ingest_manager/common/services/datasources_to_agent_inputs.ts
@@ -24,7 +24,9 @@ export const storedDatasourcesToAgentInputs = (
         id: datasource.id || datasource.name,
         name: datasource.name,
         type: input.type,
-        dataset: { namespace: datasource.namespace || 'default' },
+        dataset: {
+          namespace: datasource.namespace || 'default',
+        },
         use_output: DEFAULT_OUTPUT.name,
         ...Object.entries(input.config || {}).reduce((acc, [key, { value }]) => {
           acc[key] = value;
@@ -35,7 +37,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;
@@ -50,9 +52,11 @@ export const storedDatasourcesToAgentInputs = (
       };
 
       if (datasource.package) {
-        fullInput.package = {
-          name: datasource.package.name,
-          version: datasource.package.version,
+        fullInput.meta = {
+          package: {
+            name: datasource.package.name,
+            version: datasource.package.version,
+          },
         };
       }
 
diff --git a/x-pack/plugins/ingest_manager/common/services/package_to_config.test.ts b/x-pack/plugins/ingest_manager/common/services/package_to_config.test.ts
index a977a1a66e059a..7739fdc3a3b611 100644
--- a/x-pack/plugins/ingest_manager/common/services/package_to_config.test.ts
+++ b/x-pack/plugins/ingest_manager/common/services/package_to_config.test.ts
@@ -34,14 +34,14 @@ describe('Ingest Manager - packageToConfig', () => {
   describe('packageToConfigDatasourceInputs', () => {
     it('returns empty array for packages with no datasources', () => {
       expect(packageToConfigDatasourceInputs(mockPackage)).toEqual([]);
-      expect(packageToConfigDatasourceInputs({ ...mockPackage, datasources: [] })).toEqual([]);
+      expect(packageToConfigDatasourceInputs({ ...mockPackage, config_templates: [] })).toEqual([]);
     });
 
     it('returns empty array for packages a datasource but no inputs', () => {
       expect(
         packageToConfigDatasourceInputs(({
           ...mockPackage,
-          datasources: [{ inputs: [] }],
+          config_templates: [{ inputs: [] }],
         } as unknown) as PackageInfo)
       ).toEqual([]);
     });
@@ -50,13 +50,13 @@ describe('Ingest Manager - packageToConfig', () => {
       expect(
         packageToConfigDatasourceInputs(({
           ...mockPackage,
-          datasources: [{ inputs: [{ type: 'foo' }] }],
+          config_templates: [{ inputs: [{ type: 'foo' }] }],
         } as unknown) as PackageInfo)
       ).toEqual([{ type: 'foo', enabled: true, streams: [] }]);
       expect(
         packageToConfigDatasourceInputs(({
           ...mockPackage,
-          datasources: [{ inputs: [{ type: 'foo' }, { type: 'bar' }] }],
+          config_templates: [{ inputs: [{ type: 'foo' }, { type: 'bar' }] }],
         } as unknown) as PackageInfo)
       ).toEqual([
         { type: 'foo', enabled: true, streams: [] },
@@ -68,12 +68,14 @@ describe('Ingest Manager - packageToConfig', () => {
       expect(
         packageToConfigDatasourceInputs(({
           ...mockPackage,
-          datasources: [
+          datasets: [
+            { type: 'logs', name: 'foo', streams: [{ input: 'foo' }] },
+            { type: 'logs', name: 'bar', streams: [{ input: 'bar' }] },
+            { type: 'logs', name: 'bar2', streams: [{ input: 'bar' }] },
+          ],
+          config_templates: [
             {
-              inputs: [
-                { type: 'foo', streams: [{ dataset: 'foo' }] },
-                { type: 'bar', streams: [{ dataset: 'bar' }, { dataset: 'bar2' }] },
-              ],
+              inputs: [{ type: 'foo' }, { type: 'bar' }],
             },
           ],
         } as unknown) as PackageInfo)
@@ -81,14 +83,14 @@ describe('Ingest Manager - packageToConfig', () => {
         {
           type: 'foo',
           enabled: true,
-          streams: [{ id: 'foo-foo', enabled: true, dataset: 'foo' }],
+          streams: [{ id: 'foo-foo', enabled: true, dataset: { name: 'foo', type: 'logs' } }],
         },
         {
           type: 'bar',
           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', type: 'logs' } },
+            { id: 'bar-bar2', enabled: true, dataset: { name: 'bar2', type: 'logs' } },
           ],
         },
       ]);
@@ -98,31 +100,38 @@ describe('Ingest Manager - packageToConfig', () => {
       expect(
         packageToConfigDatasourceInputs(({
           ...mockPackage,
-          datasources: [
+          datasets: [
             {
-              inputs: [
+              type: 'logs',
+              name: 'foo',
+              streams: [{ input: 'foo', vars: [{ default: 'foo-var-value', name: 'var-name' }] }],
+            },
+            {
+              type: 'logs',
+              name: 'bar',
+              streams: [
                 {
-                  type: 'foo',
-                  streams: [
-                    { dataset: 'foo', vars: [{ default: 'foo-var-value', name: 'var-name' }] },
-                  ],
+                  input: 'bar',
+                  vars: [{ default: 'bar-var-value', name: 'var-name', type: 'text' }],
                 },
+              ],
+            },
+            {
+              type: 'logs',
+              name: 'bar2',
+              streams: [
                 {
-                  type: 'bar',
-                  streams: [
-                    {
-                      dataset: 'bar',
-                      vars: [{ default: 'bar-var-value', name: 'var-name', type: 'text' }],
-                    },
-                    {
-                      dataset: 'bar2',
-                      vars: [{ default: 'bar2-var-value', name: 'var-name', type: 'yaml' }],
-                    },
-                  ],
+                  input: 'bar',
+                  vars: [{ default: 'bar2-var-value', name: 'var-name', type: 'yaml' }],
                 },
               ],
             },
           ],
+          config_templates: [
+            {
+              inputs: [{ type: 'foo' }, { type: 'bar' }],
+            },
+          ],
         } as unknown) as PackageInfo)
       ).toEqual([
         {
@@ -132,7 +141,7 @@ describe('Ingest Manager - packageToConfig', () => {
             {
               id: 'foo-foo',
               enabled: true,
-              dataset: 'foo',
+              dataset: { name: 'foo', type: 'logs' },
               vars: { 'var-name': { value: 'foo-var-value' } },
             },
           ],
@@ -144,13 +153,13 @@ describe('Ingest Manager - packageToConfig', () => {
             {
               id: 'bar-bar',
               enabled: true,
-              dataset: 'bar',
+              dataset: { name: 'bar', type: 'logs' },
               vars: { 'var-name': { type: 'text', value: 'bar-var-value' } },
             },
             {
               id: 'bar-bar2',
               enabled: true,
-              dataset: 'bar2',
+              dataset: { name: 'bar2', type: 'logs' },
               vars: { 'var-name': { type: 'yaml', value: 'bar2-var-value' } },
             },
           ],
@@ -162,7 +171,55 @@ describe('Ingest Manager - packageToConfig', () => {
       expect(
         packageToConfigDatasourceInputs(({
           ...mockPackage,
-          datasources: [
+          datasets: [
+            {
+              type: 'logs',
+              name: 'foo',
+              streams: [{ input: 'foo', vars: [{ default: 'foo-var-value', name: 'var-name' }] }],
+            },
+            {
+              type: 'logs',
+              name: 'bar',
+              streams: [
+                {
+                  input: 'bar',
+                  vars: [{ default: 'bar-var-value', name: 'var-name' }],
+                },
+              ],
+            },
+            {
+              type: 'logs',
+              name: 'bar2',
+              streams: [
+                {
+                  input: 'bar',
+                  vars: [{ default: 'bar2-var-value', name: 'var-name' }],
+                },
+              ],
+            },
+            {
+              type: 'logs',
+              name: 'disabled',
+              streams: [
+                {
+                  input: 'with-disabled-streams',
+                  enabled: false,
+                  vars: [{ multi: true, name: 'var-name' }],
+                },
+              ],
+            },
+            {
+              type: 'logs',
+              name: 'disabled2',
+              streams: [
+                {
+                  input: 'with-disabled-streams',
+                  enabled: false,
+                },
+              ],
+            },
+          ],
+          config_templates: [
             {
               inputs: [
                 {
@@ -172,9 +229,6 @@ describe('Ingest Manager - packageToConfig', () => {
                     { default: 'foo-input2-var-value', name: 'foo-input2-var-name' },
                     { name: 'foo-input3-var-name' },
                   ],
-                  streams: [
-                    { dataset: 'foo', vars: [{ default: 'foo-var-value', name: 'var-name' }] },
-                  ],
                 },
                 {
                   type: 'bar',
@@ -182,21 +236,9 @@ describe('Ingest Manager - packageToConfig', () => {
                     { default: ['value1', 'value2'], name: 'bar-input-var-name' },
                     { default: 123456, name: 'bar-input2-var-name' },
                   ],
-                  streams: [
-                    { dataset: 'bar', vars: [{ default: 'bar-var-value', name: 'var-name' }] },
-                    { dataset: 'bar2', vars: [{ default: 'bar2-var-value', name: 'var-name' }] },
-                  ],
                 },
                 {
                   type: 'with-disabled-streams',
-                  streams: [
-                    {
-                      dataset: 'disabled',
-                      enabled: false,
-                      vars: [{ multi: true, name: 'var-name' }],
-                    },
-                    { dataset: 'disabled2', enabled: false },
-                  ],
                 },
               ],
             },
@@ -215,7 +257,7 @@ describe('Ingest Manager - packageToConfig', () => {
             {
               id: 'foo-foo',
               enabled: true,
-              dataset: 'foo',
+              dataset: { name: 'foo', type: 'logs' },
               vars: {
                 'var-name': { value: 'foo-var-value' },
               },
@@ -233,7 +275,7 @@ describe('Ingest Manager - packageToConfig', () => {
             {
               id: 'bar-bar',
               enabled: true,
-              dataset: 'bar',
+              dataset: { name: 'bar', type: 'logs' },
               vars: {
                 'var-name': { value: 'bar-var-value' },
               },
@@ -241,7 +283,7 @@ describe('Ingest Manager - packageToConfig', () => {
             {
               id: 'bar-bar2',
               enabled: true,
-              dataset: 'bar2',
+              dataset: { name: 'bar2', type: 'logs' },
               vars: {
                 'var-name': { value: 'bar2-var-value' },
               },
@@ -255,7 +297,7 @@ describe('Ingest Manager - packageToConfig', () => {
             {
               id: 'with-disabled-streams-disabled',
               enabled: false,
-              dataset: 'disabled',
+              dataset: { name: 'disabled', type: 'logs' },
               vars: {
                 'var-name': { value: [] },
               },
@@ -263,7 +305,7 @@ describe('Ingest Manager - packageToConfig', () => {
             {
               id: 'with-disabled-streams-disabled2',
               enabled: false,
-              dataset: 'disabled2',
+              dataset: { name: 'disabled2', type: 'logs' },
             },
           ],
         },
@@ -328,7 +370,7 @@ describe('Ingest Manager - packageToConfig', () => {
     it('returns datasource with inputs', () => {
       const mockPackageWithDatasources = ({
         ...mockPackage,
-        datasources: [{ inputs: [{ type: 'foo' }] }],
+        config_templates: [{ inputs: [{ type: 'foo' }] }],
       } as unknown) as PackageInfo;
 
       expect(packageToConfigDatasource(mockPackageWithDatasources, '1', '2', 'ds-1')).toEqual({
diff --git a/x-pack/plugins/ingest_manager/common/services/package_to_config.ts b/x-pack/plugins/ingest_manager/common/services/package_to_config.ts
index dc0f73b47c599a..1817077e97f9c1 100644
--- a/x-pack/plugins/ingest_manager/common/services/package_to_config.ts
+++ b/x-pack/plugins/ingest_manager/common/services/package_to_config.ts
@@ -5,8 +5,9 @@
  */
 import {
   PackageInfo,
-  RegistryDatasource,
+  RegistryConfigTemplate,
   RegistryVarsEntry,
+  RegistryStream,
   Datasource,
   DatasourceConfigRecord,
   DatasourceConfigRecordEntry,
@@ -15,6 +16,29 @@ import {
   NewDatasource,
 } from '../types';
 
+const getStreamsForInputType = (
+  inputType: string,
+  packageInfo: PackageInfo
+): 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,
+          },
+        });
+      }
+    });
+  });
+
+  return streams;
+};
+
 /*
  * This service creates a datasource inputs definition from defaults provided in package info
  */
@@ -22,8 +46,10 @@ export const packageToConfigDatasourceInputs = (packageInfo: PackageInfo): Datas
   const inputs: Datasource['inputs'] = [];
 
   // Assume package will only ever ship one datasource for now
-  const packageDatasource: RegistryDatasource | null =
-    packageInfo.datasources && packageInfo.datasources[0] ? packageInfo.datasources[0] : null;
+  const packageDatasource: RegistryConfigTemplate | null =
+    packageInfo.config_templates && packageInfo.config_templates[0]
+      ? packageInfo.config_templates[0]
+      : null;
 
   // Create datasource input property
   if (packageDatasource?.inputs?.length) {
@@ -45,19 +71,23 @@ export const packageToConfigDatasourceInputs = (packageInfo: PackageInfo): Datas
       };
 
       // Map each package input stream into datasource input stream
-      const streams: DatasourceInputStream[] = packageInput.streams
-        ? packageInput.streams.map((packageStream) => {
-            const stream: DatasourceInputStream = {
-              id: `${packageInput.type}-${packageStream.dataset}`,
-              enabled: packageStream.enabled === false ? false : true,
-              dataset: packageStream.dataset,
-            };
-            if (packageStream.vars && packageStream.vars.length) {
-              stream.vars = packageStream.vars.reduce(varsReducer, {});
-            }
-            return stream;
-          })
-        : [];
+      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: {
+            name: packageStream.dataset.name,
+            type: packageStream.dataset.type,
+          },
+        };
+        if (packageStream.vars && packageStream.vars.length) {
+          stream.vars = packageStream.vars.reduce(varsReducer, {});
+        }
+        return stream;
+      });
 
       const input: DatasourceInput = {
         type: packageInput.type,
diff --git a/x-pack/plugins/ingest_manager/common/types/models/agent_config.ts b/x-pack/plugins/ingest_manager/common/types/models/agent_config.ts
index 36b3176ffa4151..0d7dc13af7a30b 100644
--- a/x-pack/plugins/ingest_manager/common/types/models/agent_config.ts
+++ b/x-pack/plugins/ingest_manager/common/types/models/agent_config.ts
@@ -41,7 +41,10 @@ export interface FullAgentConfigInput {
   type: string;
   dataset: { namespace: string };
   use_output: string;
-  package?: Pick<DatasourcePackage, 'name' | 'version'>;
+  meta?: {
+    package?: Pick<DatasourcePackage, 'name' | 'version'>;
+    [key: string]: unknown;
+  };
   streams: FullAgentConfigInputStream[];
   [key: string]: any;
 }
diff --git a/x-pack/plugins/ingest_manager/common/types/models/datasource.ts b/x-pack/plugins/ingest_manager/common/types/models/datasource.ts
index aa92b90a6caec2..aae65bb0039959 100644
--- a/x-pack/plugins/ingest_manager/common/types/models/datasource.ts
+++ b/x-pack/plugins/ingest_manager/common/types/models/datasource.ts
@@ -20,7 +20,10 @@ export type DatasourceConfigRecord = Record<string, DatasourceConfigRecordEntry>
 export interface NewDatasourceInputStream {
   id: string;
   enabled: boolean;
-  dataset: string;
+  dataset: {
+    name: string;
+    type: string;
+  };
   processors?: string[];
   config?: DatasourceConfigRecord;
   vars?: DatasourceConfigRecord;
diff --git a/x-pack/plugins/ingest_manager/common/types/models/epm.ts b/x-pack/plugins/ingest_manager/common/types/models/epm.ts
index 599165d2bfd981..3ae0628fa695d2 100644
--- a/x-pack/plugins/ingest_manager/common/types/models/epm.ts
+++ b/x-pack/plugins/ingest_manager/common/types/models/epm.ts
@@ -60,7 +60,7 @@ export interface RegistryPackage {
   internal?: boolean;
   format_version: string;
   datasets?: Dataset[];
-  datasources?: RegistryDatasource[];
+  config_templates?: RegistryConfigTemplate[];
   download: string;
   path: string;
 }
@@ -74,7 +74,7 @@ interface RegistryImage {
   size?: string;
   type?: string;
 }
-export interface RegistryDatasource {
+export interface RegistryConfigTemplate {
   name: string;
   title: string;
   description: string;
@@ -86,17 +86,15 @@ export interface RegistryInput {
   title: string;
   description?: string;
   vars?: RegistryVarsEntry[];
-  streams: RegistryStream[];
 }
 
 export interface RegistryStream {
   input: string;
-  dataset: string;
   title: string;
   description?: string;
   enabled?: boolean;
   vars?: RegistryVarsEntry[];
-  template?: string;
+  template_path: string;
 }
 
 export type RequirementVersion = string;
@@ -122,7 +120,7 @@ export type RegistrySearchResult = Pick<
   | 'download'
   | 'path'
   | 'datasets'
-  | 'datasources'
+  | 'config_templates'
 >;
 
 export type ScreenshotItem = RegistryImage;
@@ -169,15 +167,14 @@ export type ElasticsearchAssetTypeToParts = Record<
 >;
 
 export interface Dataset {
+  type: string;
+  name: string;
   title: string;
-  path: string;
-  id: string;
   release: string;
-  ingest_pipeline: string;
-  vars?: RegistryVarsEntry[];
-  type: string;
   streams?: RegistryStream[];
   package: string;
+  path: string;
+  ingest_pipeline: string;
 }
 
 // EPR types this as `[]map[string]interface{}`
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_panel.tsx b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_panel.tsx
index db704d8b1d0f3c..6f6fa5aaa7f3e8 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_panel.tsx
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_panel.tsx
@@ -19,7 +19,12 @@ import {
   EuiSpacer,
   EuiIconTip,
 } from '@elastic/eui';
-import { DatasourceInput, DatasourceInputStream, RegistryInput } from '../../../../types';
+import {
+  DatasourceInput,
+  DatasourceInputStream,
+  RegistryInput,
+  RegistryStream,
+} from '../../../../types';
 import { DatasourceInputValidationResults, validationHasErrors } from '../services';
 import { DatasourceInputConfig } from './datasource_input_config';
 import { DatasourceInputStreamConfig } from './datasource_input_stream_config';
@@ -32,12 +37,14 @@ const FlushHorizontalRule = styled(EuiHorizontalRule)`
 
 export const DatasourceInputPanel: React.FunctionComponent<{
   packageInput: RegistryInput;
+  packageInputStreams: Array<RegistryStream & { dataset: { name: string } }>;
   datasourceInput: DatasourceInput;
   updateDatasourceInput: (updatedInput: Partial<DatasourceInput>) => void;
   inputValidationResults: DatasourceInputValidationResults;
   forceShowErrors?: boolean;
 }> = ({
   packageInput,
+  packageInputStreams,
   datasourceInput,
   updateDatasourceInput,
   inputValidationResults,
@@ -111,7 +118,7 @@ export const DatasourceInputPanel: React.FunctionComponent<{
                         </strong>
                       </EuiTextColor>
                     ),
-                    total: packageInput.streams.length,
+                    total: packageInputStreams.length,
                   }}
                 />
               </EuiText>
@@ -168,18 +175,18 @@ export const DatasourceInputPanel: React.FunctionComponent<{
       {/* Per-stream configuration */}
       {isShowingStreams ? (
         <EuiFlexGroup direction="column">
-          {packageInput.streams.map((packageInputStream) => {
+          {packageInputStreams.map((packageInputStream) => {
             const datasourceInputStream = datasourceInput.streams.find(
-              (stream) => stream.dataset === packageInputStream.dataset
+              (stream) => stream.dataset.name === packageInputStream.dataset.name
             );
             return datasourceInputStream ? (
-              <EuiFlexItem key={packageInputStream.dataset}>
+              <EuiFlexItem key={packageInputStream.dataset.name}>
                 <DatasourceInputStreamConfig
                   packageInputStream={packageInputStream}
                   datasourceInputStream={datasourceInputStream}
                   updateDatasourceInputStream={(updatedStream: Partial<DatasourceInputStream>) => {
                     const indexOfUpdatedStream = datasourceInput.streams.findIndex(
-                      (stream) => stream.dataset === packageInputStream.dataset
+                      (stream) => stream.dataset.name === packageInputStream.dataset.name
                     );
                     const newStreams = [...datasourceInput.streams];
                     newStreams[indexOfUpdatedStream] = {
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_stream_config.tsx b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_stream_config.tsx
index 978ad83cd5c3c9..f697ef736ef705 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_stream_config.tsx
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_stream_config.tsx
@@ -60,7 +60,7 @@ export const DatasourceInputStreamConfig: React.FunctionComponent<{
             <EuiFlexGroup alignItems="center" gutterSize="s">
               <EuiFlexItem grow={false}>
                 <EuiTextColor color={hasErrors ? 'danger' : 'default'}>
-                  {packageInputStream.title || packageInputStream.dataset}
+                  {packageInputStream.title}
                 </EuiTextColor>
               </EuiFlexItem>
               {hasErrors ? (
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.test.ts b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.test.ts
index 67cde2dec3a56e..64facf01d474a7 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.test.ts
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.test.ts
@@ -7,7 +7,7 @@ import {
   PackageInfo,
   InstallationStatus,
   NewDatasource,
-  RegistryDatasource,
+  RegistryConfigTemplate,
 } from '../../../../types';
 import { validateDatasource, validationHasErrors } from './validate_datasource';
 
@@ -32,7 +32,65 @@ describe('Ingest Manager - validateDatasource()', () => {
       },
     },
     status: InstallationStatus.notInstalled,
-    datasources: [
+    datasets: [
+      {
+        name: 'foo',
+        streams: [
+          {
+            input: 'foo',
+            title: 'Foo',
+            vars: [{ name: 'var-name', type: 'yaml' }],
+          },
+        ],
+      },
+      {
+        name: 'bar',
+        streams: [
+          {
+            input: 'bar',
+            title: 'Bar',
+            vars: [{ name: 'var-name', type: 'yaml', required: true }],
+          },
+          {
+            input: 'with-no-stream-vars',
+            title: 'Bar stream no vars',
+            enabled: true,
+          },
+        ],
+      },
+      {
+        name: 'bar2',
+        streams: [
+          {
+            input: 'bar',
+            title: 'Bar 2',
+            vars: [{ default: 'bar2-var-value', name: 'var-name', type: 'text' }],
+          },
+        ],
+      },
+      {
+        name: 'disabled',
+        streams: [
+          {
+            input: 'with-disabled-streams',
+            title: 'Disabled',
+            enabled: false,
+            vars: [{ multi: true, required: true, name: 'var-name', type: 'text' }],
+          },
+        ],
+      },
+      {
+        name: 'disabled2',
+        streams: [
+          {
+            input: 'with-disabled-streams',
+            title: 'Disabled 2',
+            enabled: false,
+          },
+        ],
+      },
+    ],
+    config_templates: [
       {
         name: 'datasource1',
         title: 'Datasource 1',
@@ -51,14 +109,6 @@ describe('Ingest Manager - validateDatasource()', () => {
               },
               { name: 'foo-input3-var-name', type: 'text', required: true, multi: true },
             ],
-            streams: [
-              {
-                dataset: 'foo',
-                input: 'foo',
-                title: 'Foo',
-                vars: [{ name: 'var-name', type: 'yaml' }],
-              },
-            ],
           },
           {
             type: 'bar',
@@ -72,51 +122,19 @@ describe('Ingest Manager - validateDatasource()', () => {
               },
               { name: 'bar-input2-var-name', required: true, type: 'text' },
             ],
-            streams: [
-              {
-                dataset: 'bar',
-                input: 'bar',
-                title: 'Bar',
-                vars: [{ name: 'var-name', type: 'yaml', required: true }],
-              },
-              {
-                dataset: 'bar2',
-                input: 'bar2',
-                title: 'Bar 2',
-                vars: [{ default: 'bar2-var-value', name: 'var-name', type: 'text' }],
-              },
-            ],
           },
           {
             type: 'with-no-config-or-streams',
             title: 'With no config or streams',
-            streams: [],
           },
           {
             type: 'with-disabled-streams',
             title: 'With disabled streams',
-            streams: [
-              {
-                dataset: 'disabled',
-                input: 'disabled',
-                title: 'Disabled',
-                enabled: false,
-                vars: [{ multi: true, required: true, name: 'var-name', type: 'text' }],
-              },
-              { dataset: 'disabled2', input: 'disabled2', title: 'Disabled 2', enabled: false },
-            ],
           },
           {
             type: 'with-no-stream-vars',
             enabled: true,
             vars: [{ required: true, name: 'var-name', type: 'text' }],
-            streams: [
-              {
-                id: 'with-no-stream-vars-bar',
-                dataset: 'bar',
-                enabled: true,
-              },
-            ],
           },
         ],
       },
@@ -140,7 +158,7 @@ describe('Ingest Manager - validateDatasource()', () => {
         streams: [
           {
             id: 'foo-foo',
-            dataset: 'foo',
+            dataset: { name: 'foo', type: 'logs' },
             enabled: true,
             vars: { 'var-name': { value: 'test_yaml: value', type: 'yaml' } },
           },
@@ -156,13 +174,13 @@ describe('Ingest Manager - validateDatasource()', () => {
         streams: [
           {
             id: 'bar-bar',
-            dataset: 'bar',
+            dataset: { name: 'bar', type: 'logs' },
             enabled: true,
             vars: { 'var-name': { value: 'test_yaml: value', type: 'yaml' } },
           },
           {
             id: 'bar-bar2',
-            dataset: 'bar2',
+            dataset: { name: 'bar2', type: 'logs' },
             enabled: true,
             vars: { 'var-name': { value: undefined, type: 'text' } },
           },
@@ -179,13 +197,13 @@ describe('Ingest Manager - validateDatasource()', () => {
         streams: [
           {
             id: 'with-disabled-streams-disabled',
-            dataset: 'disabled',
+            dataset: { name: 'disabled', type: 'logs' },
             enabled: false,
             vars: { 'var-name': { value: undefined, type: 'text' } },
           },
           {
             id: 'with-disabled-streams-disabled-without-vars',
-            dataset: 'disabled2',
+            dataset: { name: 'disabled2', type: 'logs' },
             enabled: false,
           },
         ],
@@ -199,7 +217,7 @@ describe('Ingest Manager - validateDatasource()', () => {
         streams: [
           {
             id: 'with-no-stream-vars-bar',
-            dataset: 'bar',
+            dataset: { name: 'bar', type: 'logs' },
             enabled: true,
           },
         ],
@@ -222,7 +240,7 @@ describe('Ingest Manager - validateDatasource()', () => {
         streams: [
           {
             id: 'foo-foo',
-            dataset: 'foo',
+            dataset: { name: 'foo', type: 'logs' },
             enabled: true,
             vars: { 'var-name': { value: 'invalidyaml: test\n foo bar:', type: 'yaml' } },
           },
@@ -238,13 +256,13 @@ describe('Ingest Manager - validateDatasource()', () => {
         streams: [
           {
             id: 'bar-bar',
-            dataset: 'bar',
+            dataset: { name: 'bar', type: 'logs' },
             enabled: true,
             vars: { 'var-name': { value: '    \n\n', type: 'yaml' } },
           },
           {
             id: 'bar-bar2',
-            dataset: 'bar2',
+            dataset: { name: 'bar2', type: 'logs' },
             enabled: true,
             vars: { 'var-name': { value: undefined, type: 'text' } },
           },
@@ -261,7 +279,7 @@ describe('Ingest Manager - validateDatasource()', () => {
         streams: [
           {
             id: 'with-disabled-streams-disabled',
-            dataset: 'disabled',
+            dataset: { name: 'disabled', type: 'logs' },
             enabled: false,
             vars: {
               'var-name': {
@@ -272,7 +290,7 @@ describe('Ingest Manager - validateDatasource()', () => {
           },
           {
             id: 'with-disabled-streams-disabled-without-vars',
-            dataset: 'disabled2',
+            dataset: { name: 'disabled2', type: 'logs' },
             enabled: false,
           },
         ],
@@ -286,7 +304,7 @@ describe('Ingest Manager - validateDatasource()', () => {
         streams: [
           {
             id: 'with-no-stream-vars-bar',
-            dataset: 'bar',
+            dataset: { name: 'bar', type: 'logs' },
             enabled: true,
           },
         ],
@@ -435,7 +453,7 @@ describe('Ingest Manager - validateDatasource()', () => {
     expect(
       validateDatasource(validDatasource, {
         ...mockPackage,
-        datasources: undefined,
+        config_templates: undefined,
       })
     ).toEqual({
       name: null,
@@ -445,7 +463,7 @@ describe('Ingest Manager - validateDatasource()', () => {
     expect(
       validateDatasource(validDatasource, {
         ...mockPackage,
-        datasources: [],
+        config_templates: [],
       })
     ).toEqual({
       name: null,
@@ -458,7 +476,7 @@ describe('Ingest Manager - validateDatasource()', () => {
     expect(
       validateDatasource(validDatasource, {
         ...mockPackage,
-        datasources: [{} as RegistryDatasource],
+        config_templates: [{} as RegistryConfigTemplate],
       })
     ).toEqual({
       name: null,
@@ -468,7 +486,7 @@ describe('Ingest Manager - validateDatasource()', () => {
     expect(
       validateDatasource(validDatasource, {
         ...mockPackage,
-        datasources: [({ inputs: [] } as unknown) as RegistryDatasource],
+        config_templates: [({ inputs: [] } as unknown) as RegistryConfigTemplate],
       })
     ).toEqual({
       name: null,
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.ts b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.ts
index 5b4cfe170a4786..30dca4a5fbf816 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.ts
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.ts
@@ -13,6 +13,7 @@ import {
   DatasourceConfigRecordEntry,
   PackageInfo,
   RegistryInput,
+  RegistryStream,
   RegistryVarsEntry,
 } from '../../../../types';
 
@@ -57,11 +58,11 @@ export const validateDatasource = (
   }
 
   if (
-    !packageInfo.datasources ||
-    packageInfo.datasources.length === 0 ||
-    !packageInfo.datasources[0] ||
-    !packageInfo.datasources[0].inputs ||
-    packageInfo.datasources[0].inputs.length === 0
+    !packageInfo.config_templates ||
+    packageInfo.config_templates.length === 0 ||
+    !packageInfo.config_templates[0] ||
+    !packageInfo.config_templates[0].inputs ||
+    packageInfo.config_templates[0].inputs.length === 0
   ) {
     validationResults.inputs = null;
     return validationResults;
@@ -70,11 +71,18 @@ export const validateDatasource = (
   const registryInputsByType: Record<
     string,
     RegistryInput
-  > = packageInfo.datasources[0].inputs.reduce((inputs, registryInput) => {
+  > = packageInfo.config_templates[0].inputs.reduce((inputs, registryInput) => {
     inputs[registryInput.type] = registryInput;
     return inputs;
   }, {} as Record<string, RegistryInput>);
 
+  const registryStreamsByDataset: Record<string, RegistryStream[]> = (
+    packageInfo.datasets || []
+  ).reduce((datasets, registryDataset) => {
+    datasets[registryDataset.name] = registryDataset.streams || [];
+    return datasets;
+  }, {} as Record<string, RegistryStream[]>);
+
   // Validate each datasource input with either its own config fields or streams
   datasource.inputs.forEach((input) => {
     if (!input.vars && !input.streams) {
@@ -116,8 +124,8 @@ export const validateDatasource = (
         if (stream.vars) {
           const streamVarsByName = (
             (
-              registryInputsByType[input.type].streams.find(
-                (registryStream) => registryStream.dataset === stream.dataset
+              registryStreamsByDataset[stream.dataset.name].find(
+                (registryStream) => registryStream.input === input.type
               ) || {}
             ).vars || []
           ).reduce((vars, registryVar) => {
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/step_configure_datasource.tsx b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/step_configure_datasource.tsx
index 5499ac287ff051..c319dbc08b2c6b 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/step_configure_datasource.tsx
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/step_configure_datasource.tsx
@@ -7,12 +7,34 @@ import React from 'react';
 import { FormattedMessage } from '@kbn/i18n/react';
 import { EuiPanel, EuiFlexGroup, EuiFlexItem, EuiSpacer, EuiCallOut } from '@elastic/eui';
 import { i18n } from '@kbn/i18n';
-import { PackageInfo, NewDatasource, DatasourceInput } from '../../../types';
+import { PackageInfo, RegistryStream, NewDatasource, DatasourceInput } from '../../../types';
 import { Loading } from '../../../components';
 import { DatasourceValidationResults, validationHasErrors } from './services';
 import { DatasourceInputPanel, CustomConfigureDatasource } from './components';
 import { CreateDatasourceFrom } from './types';
 
+const findStreamsForInputType = (
+  inputType: string,
+  packageInfo: PackageInfo
+): Array<RegistryStream & { dataset: { name: string } }> => {
+  const streams: Array<RegistryStream & { dataset: { name: string } }> = [];
+
+  (packageInfo.datasets || []).forEach((dataset) => {
+    (dataset.streams || []).forEach((stream) => {
+      if (stream.input === inputType) {
+        streams.push({
+          ...stream,
+          dataset: {
+            name: dataset.name,
+          },
+        });
+      }
+    });
+  });
+
+  return streams;
+};
+
 export const StepConfigureDatasource: React.FunctionComponent<{
   from?: CreateDatasourceFrom;
   packageInfo: PackageInfo;
@@ -35,19 +57,21 @@ export const StepConfigureDatasource: React.FunctionComponent<{
   // Configure inputs (and their streams)
   // Assume packages only export one datasource for now
   const renderConfigureInputs = () =>
-    packageInfo.datasources &&
-    packageInfo.datasources[0] &&
-    packageInfo.datasources[0].inputs &&
-    packageInfo.datasources[0].inputs.length ? (
+    packageInfo.config_templates &&
+    packageInfo.config_templates[0] &&
+    packageInfo.config_templates[0].inputs &&
+    packageInfo.config_templates[0].inputs.length ? (
       <EuiFlexGroup direction="column">
-        {packageInfo.datasources[0].inputs.map((packageInput) => {
+        {packageInfo.config_templates[0].inputs.map((packageInput) => {
           const datasourceInput = datasource.inputs.find(
             (input) => input.type === packageInput.type
           );
+          const packageInputStreams = findStreamsForInputType(packageInput.type, packageInfo);
           return datasourceInput ? (
             <EuiFlexItem key={packageInput.type}>
               <DatasourceInputPanel
                 packageInput={packageInput}
+                packageInputStreams={packageInputStreams}
                 datasourceInput={datasourceInput}
                 updateDatasourceInput={(updatedInput: Partial<DatasourceInput>) => {
                   const indexOfUpdatedInput = datasource.inputs.findIndex(
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/types/index.ts b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/types/index.ts
index 1a4c6a8a86f6ed..90315f38fd476b 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/types/index.ts
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/types/index.ts
@@ -81,7 +81,7 @@ export {
   RegistryVarsEntry,
   RegistryInput,
   RegistryStream,
-  RegistryDatasource,
+  RegistryConfigTemplate,
   PackageList,
   PackageListItem,
   PackagesGroupedByStatus,
diff --git a/x-pack/plugins/ingest_manager/server/saved_objects/index.ts b/x-pack/plugins/ingest_manager/server/saved_objects/index.ts
index 9afc06c193a3b1..d00e403849a768 100644
--- a/x-pack/plugins/ingest_manager/server/saved_objects/index.ts
+++ b/x-pack/plugins/ingest_manager/server/saved_objects/index.ts
@@ -212,7 +212,12 @@ const savedObjectTypes: { [key: string]: SavedObjectsType } = {
               properties: {
                 id: { type: 'keyword' },
                 enabled: { type: 'boolean' },
-                dataset: { type: 'keyword' },
+                dataset: {
+                  properties: {
+                    name: { type: 'keyword' },
+                    type: { type: 'keyword' },
+                  },
+                },
                 processors: { type: 'keyword' },
                 config: { type: 'flattened' },
                 agent_stream: { type: 'flattened' },
diff --git a/x-pack/plugins/ingest_manager/server/saved_objects/migrations/datasources_v790.ts b/x-pack/plugins/ingest_manager/server/saved_objects/migrations/datasources_v790.ts
index 0d1fb6f21a1ae7..d1e4b29daefc60 100644
--- a/x-pack/plugins/ingest_manager/server/saved_objects/migrations/datasources_v790.ts
+++ b/x-pack/plugins/ingest_manager/server/saved_objects/migrations/datasources_v790.ts
@@ -6,12 +6,22 @@
 
 import { SavedObjectMigrationFn } from 'kibana/server';
 import { cloneDeep } from 'lodash';
-import { Datasource } from '../../types/models';
+import { Datasource, DatasourceInput, DatasourceInputStream } from '../../types';
 
 type Pre790Datasource = Exclude<
   Datasource,
   'created_at' | 'created_by' | 'updated_at' | 'updated_by'
->;
+> & {
+  inputs: Array<
+    Exclude<DatasourceInput, 'streams'> & {
+      streams: Array<
+        Exclude<DatasourceInputStream, 'dataset'> & {
+          dataset: string;
+        }
+      >;
+    }
+  >;
+};
 
 export const migrateDatasourcesToV790: SavedObjectMigrationFn<Pre790Datasource, Datasource> = (
   doc
@@ -23,6 +33,11 @@ export const migrateDatasourcesToV790: SavedObjectMigrationFn<Pre790Datasource,
   updatedDatasource.attributes.created_at = updatedDatasource?.updated_at ?? defDate;
   updatedDatasource.attributes.updated_by = 'system';
   updatedDatasource.attributes.updated_at = updatedDatasource?.updated_at ?? defDate;
+  updatedDatasource.attributes.inputs.forEach((input) => {
+    input.streams.forEach((stream) => {
+      stream.dataset = { name: (stream.dataset as unknown) as string, type: '' };
+    });
+  });
 
   return updatedDatasource;
 };
diff --git a/x-pack/plugins/ingest_manager/server/services/datasource.test.ts b/x-pack/plugins/ingest_manager/server/services/datasource.test.ts
index 3682ae6d1167b9..8d98e41c8ae690 100644
--- a/x-pack/plugins/ingest_manager/server/services/datasource.test.ts
+++ b/x-pack/plugins/ingest_manager/server/services/datasource.test.ts
@@ -7,33 +7,51 @@
 import { datasourceService } from './datasource';
 import { PackageInfo } from '../types';
 
-const TEMPLATE = `
+async function mockedGetAssetsData(_a: any, _b: any, dataset: string) {
+  if (dataset === 'dataset1') {
+    return [
+      {
+        buffer: Buffer.from(`
 type: log
 metricset: ["dataset1"]
 paths:
 {{#each paths}}
 - {{this}}
 {{/each}}
-`;
+`),
+      },
+    ];
+  }
+  return [];
+}
+
+jest.mock('./epm/packages/assets', () => {
+  return {
+    getAssetsData: mockedGetAssetsData,
+  };
+});
+
+jest.mock('./epm/registry', () => {
+  return {
+    fetchInfo: () => ({}),
+  };
+});
 
 describe('Datasource service', () => {
   describe('assignPackageStream', () => {
     it('should work with config variables from the stream', async () => {
       const inputs = await datasourceService.assignPackageStream(
         ({
-          datasources: [
+          datasets: [
             {
-              inputs: [
-                {
-                  type: 'log',
-                  streams: [
-                    {
-                      dataset: 'package.dataset1',
-                      template: TEMPLATE,
-                    },
-                  ],
-                },
-              ],
+              type: 'logs',
+              name: 'package.dataset1',
+              streams: [{ input: 'log', template_path: 'some_template_path.yml' }],
+            },
+          ],
+          config_templates: [
+            {
+              inputs: [{ type: 'log' }],
             },
           ],
         } as unknown) as PackageInfo,
@@ -44,7 +62,7 @@ describe('Datasource service', () => {
             streams: [
               {
                 id: 'dataset01',
-                dataset: 'package.dataset1',
+                dataset: { name: 'package.dataset1', type: 'logs' },
                 enabled: true,
                 vars: {
                   paths: {
@@ -64,7 +82,7 @@ describe('Datasource service', () => {
           streams: [
             {
               id: 'dataset01',
-              dataset: 'package.dataset1',
+              dataset: { name: 'package.dataset1', type: 'logs' },
               enabled: true,
               vars: {
                 paths: {
@@ -85,19 +103,16 @@ describe('Datasource service', () => {
     it('should work with config variables at the input level', async () => {
       const inputs = await datasourceService.assignPackageStream(
         ({
-          datasources: [
+          datasets: [
             {
-              inputs: [
-                {
-                  type: 'log',
-                  streams: [
-                    {
-                      dataset: 'package.dataset1',
-                      template: TEMPLATE,
-                    },
-                  ],
-                },
-              ],
+              name: 'package.dataset1',
+              type: 'logs',
+              streams: [{ input: 'log', template_path: 'some_template_path.yml' }],
+            },
+          ],
+          config_templates: [
+            {
+              inputs: [{ type: 'log' }],
             },
           ],
         } as unknown) as PackageInfo,
@@ -113,7 +128,7 @@ describe('Datasource service', () => {
             streams: [
               {
                 id: 'dataset01',
-                dataset: 'package.dataset1',
+                dataset: { name: 'package.dataset1', type: 'logs' },
                 enabled: true,
               },
             ],
@@ -133,7 +148,7 @@ describe('Datasource service', () => {
           streams: [
             {
               id: 'dataset01',
-              dataset: 'package.dataset1',
+              dataset: { name: 'package.dataset1', type: 'logs' },
               enabled: true,
               agent_stream: {
                 metricset: ['dataset1'],
diff --git a/x-pack/plugins/ingest_manager/server/services/datasource.ts b/x-pack/plugins/ingest_manager/server/services/datasource.ts
index f3f460d2a74206..fecec0c463459c 100644
--- a/x-pack/plugins/ingest_manager/server/services/datasource.ts
+++ b/x-pack/plugins/ingest_manager/server/services/datasource.ts
@@ -13,10 +13,18 @@ import {
   PackageInfo,
 } from '../../common';
 import { DATASOURCE_SAVED_OBJECT_TYPE } from '../constants';
-import { NewDatasource, Datasource, ListWithKuery, DatasourceSOAttributes } from '../types';
+import {
+  NewDatasource,
+  Datasource,
+  ListWithKuery,
+  DatasourceSOAttributes,
+  RegistryPackage,
+} from '../types';
 import { agentConfigService } from './agent_config';
-import { getPackageInfo, getInstallation } from './epm/packages';
 import { outputService } from './output';
+import * as Registry from './epm/registry';
+import { getPackageInfo, getInstallation } from './epm/packages';
+import { getAssetsData } from './epm/packages/assets';
 import { createStream } from './epm/agent/agent';
 
 const SAVED_OBJECT_TYPE = DATASOURCE_SAVED_OBJECT_TYPE;
@@ -251,15 +259,22 @@ class DatasourceService {
     pkgInfo: PackageInfo,
     inputs: DatasourceInput[]
   ): Promise<DatasourceInput[]> {
-    const inputsPromises = inputs.map((input) => _assignPackageStreamToInput(pkgInfo, input));
+    const registryPkgInfo = await Registry.fetchInfo(pkgInfo.name, pkgInfo.version);
+    const inputsPromises = inputs.map((input) =>
+      _assignPackageStreamToInput(registryPkgInfo, pkgInfo, input)
+    );
 
     return Promise.all(inputsPromises);
   }
 }
 
-async function _assignPackageStreamToInput(pkgInfo: PackageInfo, input: DatasourceInput) {
+async function _assignPackageStreamToInput(
+  registryPkgInfo: RegistryPackage,
+  pkgInfo: PackageInfo,
+  input: DatasourceInput
+) {
   const streamsPromises = input.streams.map((stream) =>
-    _assignPackageStreamToStream(pkgInfo, input, stream)
+    _assignPackageStreamToStream(registryPkgInfo, pkgInfo, input, stream)
   );
 
   const streams = await Promise.all(streamsPromises);
@@ -267,6 +282,7 @@ async function _assignPackageStreamToInput(pkgInfo: PackageInfo, input: Datasour
 }
 
 async function _assignPackageStreamToStream(
+  registryPkgInfo: RegistryPackage,
   pkgInfo: PackageInfo,
   input: DatasourceInput,
   stream: DatasourceInputStream
@@ -274,32 +290,46 @@ async function _assignPackageStreamToStream(
   if (!stream.enabled) {
     return { ...stream, agent_stream: undefined };
   }
-  const dataset = getDataset(stream.dataset);
-  const datasource = pkgInfo.datasources?.[0];
-  if (!datasource) {
-    throw new Error('Stream template not found, no datasource');
+  const datasetPath = getDataset(stream.dataset.name);
+  const packageDatasets = pkgInfo.datasets;
+  if (!packageDatasets) {
+    throw new Error('Stream template not found, no datasets');
   }
 
-  const inputFromPkg = datasource.inputs.find((pkgInput) => pkgInput.type === input.type);
-  if (!inputFromPkg) {
-    throw new Error(`Stream template not found, unable to found input ${input.type}`);
+  const packageDataset = packageDatasets.find(
+    (pkgDataset) => pkgDataset.name === stream.dataset.name
+  );
+  if (!packageDataset) {
+    throw new Error(`Stream template not found, unable to find dataset ${datasetPath}`);
   }
 
-  const streamFromPkg = inputFromPkg.streams.find(
-    (pkgStream) => pkgStream.dataset === stream.dataset
+  const streamFromPkg = (packageDataset.streams || []).find(
+    (pkgStream) => pkgStream.input === input.type
   );
   if (!streamFromPkg) {
-    throw new Error(`Stream template not found, unable to found stream ${stream.dataset}`);
+    throw new Error(`Stream template not found, unable to find stream for input ${input.type}`);
+  }
+
+  if (!streamFromPkg.template_path) {
+    throw new Error(`Stream template path not found for dataset ${datasetPath}`);
   }
 
-  if (!streamFromPkg.template) {
-    throw new Error(`Stream template not found for dataset ${dataset}`);
+  const [pkgStream] = await getAssetsData(
+    registryPkgInfo,
+    (path: string) => path.endsWith(streamFromPkg.template_path),
+    datasetPath
+  );
+
+  if (!pkgStream || !pkgStream.buffer) {
+    throw new Error(
+      `Unable to load stream template ${streamFromPkg.template_path} for dataset ${datasetPath}`
+    );
   }
 
   const yaml = createStream(
     // Populate template variables from input vars and stream vars
     Object.assign({}, input.vars, stream.vars),
-    streamFromPkg.template
+    pkgStream.buffer.toString()
   );
 
   stream.agent_stream = yaml;
diff --git a/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/index.test.ts b/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/index.test.ts
index 851a3bc2dd7206..bdd8883ea29c24 100644
--- a/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/index.test.ts
+++ b/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/index.test.ts
@@ -9,7 +9,7 @@ import { getDatasetAssetBaseName } from './index';
 
 test('getBaseName', () => {
   const dataset: Dataset = {
-    id: 'nginx.access',
+    name: 'nginx.access',
     title: 'Nginx Acess Logs',
     release: 'beta',
     type: 'logs',
diff --git a/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/index.ts b/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/index.ts
index e00b9db71db10b..0cb09ba054bf10 100644
--- a/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/index.ts
+++ b/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/index.ts
@@ -11,5 +11,5 @@ import { Dataset } from '../../../types';
  * {type}-{id}
  */
 export function getDatasetAssetBaseName(dataset: Dataset): string {
-  return `${dataset.type}-${dataset.id}`;
+  return `${dataset.type}-${dataset.name}`;
 }
diff --git a/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/ingest_pipelines.test.ts b/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/ingest_pipelines.test.ts
index c3b135993105e5..36a19c512a8b48 100644
--- a/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/ingest_pipelines.test.ts
+++ b/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/ingest_pipelines.test.ts
@@ -107,7 +107,7 @@ test('a yml-format pipeline with no pipeline references stays unchanged', () =>
 
 test('getPipelineNameForInstallation gets correct name', () => {
   const dataset: Dataset = {
-    id: 'coredns.log',
+    name: 'coredns.log',
     title: 'CoreDNS logs',
     release: 'ga',
     type: 'logs',
@@ -127,8 +127,10 @@ test('getPipelineNameForInstallation gets correct name', () => {
     dataset,
     packageVersion,
   });
-  expect(pipelineEntryNameForInstallation).toBe(`${dataset.type}-${dataset.id}-${packageVersion}`);
+  expect(pipelineEntryNameForInstallation).toBe(
+    `${dataset.type}-${dataset.name}-${packageVersion}`
+  );
   expect(pipelineRefNameForInstallation).toBe(
-    `${dataset.type}-${dataset.id}-${packageVersion}-${pipelineRefName}`
+    `${dataset.type}-${dataset.name}-${packageVersion}-${pipelineRefName}`
   );
 });
diff --git a/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/install.ts b/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/install.ts
index 11543fe73886f2..0865ee5d59e57a 100644
--- a/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/install.ts
+++ b/x-pack/plugins/ingest_manager/server/services/epm/elasticsearch/ingest_pipeline/install.ts
@@ -192,5 +192,5 @@ export const getPipelineNameForInstallation = ({
   const isPipelineEntry = pipelineName === dataset.ingest_pipeline;
   const suffix = isPipelineEntry ? '' : `-${pipelineName}`;
   // if this is the pipeline entry, don't add a suffix
-  return `${dataset.type}-${dataset.id}-${packageVersion}${suffix}`;
+  return `${dataset.type}-${dataset.name}-${packageVersion}${suffix}`;
 };
diff --git a/x-pack/plugins/ingest_manager/server/services/epm/packages/assets.ts b/x-pack/plugins/ingest_manager/server/services/epm/packages/assets.ts
index 37fcf0db671310..19a023eb2ad4c5 100644
--- a/x-pack/plugins/ingest_manager/server/services/epm/packages/assets.ts
+++ b/x-pack/plugins/ingest_manager/server/services/epm/packages/assets.ts
@@ -71,13 +71,3 @@ export async function getAssetsData(
 
   return entries;
 }
-
-export async function getAssetsDataForPackageKey(
-  { pkgName, pkgVersion }: { pkgName: string; pkgVersion: string },
-  filter = (path: string): boolean => true,
-  datasetName?: string
-): Promise<Registry.ArchiveEntry[]> {
-  const registryPkgInfo = await Registry.fetchInfo(pkgName, pkgVersion);
-
-  return getAssetsData(registryPkgInfo, filter, datasetName);
-}
diff --git a/x-pack/plugins/ingest_manager/server/types/index.tsx b/x-pack/plugins/ingest_manager/server/types/index.tsx
index d8214e95aa2a05..eedb5d86abda32 100644
--- a/x-pack/plugins/ingest_manager/server/types/index.tsx
+++ b/x-pack/plugins/ingest_manager/server/types/index.tsx
@@ -18,6 +18,8 @@ export {
   AgentAction,
   AgentActionSOAttributes,
   Datasource,
+  DatasourceInput,
+  DatasourceInputStream,
   NewDatasource,
   DatasourceSOAttributes,
   FullAgentConfigInput,
diff --git a/x-pack/plugins/ingest_manager/server/types/models/datasource.ts b/x-pack/plugins/ingest_manager/server/types/models/datasource.ts
index 3bca6d20d96a20..114986c4a486e5 100644
--- a/x-pack/plugins/ingest_manager/server/types/models/datasource.ts
+++ b/x-pack/plugins/ingest_manager/server/types/models/datasource.ts
@@ -4,7 +4,6 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 import { schema } from '@kbn/config-schema';
-export { Datasource, NewDatasource } from '../../../common';
 
 const ConfigRecordSchema = schema.recordOf(
   schema.string(),
@@ -47,7 +46,7 @@ const DatasourceBaseSchema = {
         schema.object({
           id: schema.string(),
           enabled: schema.boolean(),
-          dataset: schema.string(),
+          dataset: schema.object({ name: schema.string(), type: schema.string() }),
           processors: schema.maybe(schema.arrayOf(schema.string())),
           vars: schema.maybe(ConfigRecordSchema),
           config: schema.maybe(
diff --git a/x-pack/plugins/ingest_manager/server/types/models/output.ts b/x-pack/plugins/ingest_manager/server/types/models/output.ts
index 36b945db2cbcef..22a101ecd94b8e 100644
--- a/x-pack/plugins/ingest_manager/server/types/models/output.ts
+++ b/x-pack/plugins/ingest_manager/server/types/models/output.ts
@@ -4,7 +4,6 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 import { schema } from '@kbn/config-schema';
-export { Output, NewOutput } from '../../../common';
 
 export enum OutputType {
   Elasticsearch = 'elasticsearch',
diff --git a/x-pack/test/security_solution_endpoint/apps/endpoint/policy_details.ts b/x-pack/test/security_solution_endpoint/apps/endpoint/policy_details.ts
index b0c161ca1d0c24..f0e47c68866010 100644
--- a/x-pack/test/security_solution_endpoint/apps/endpoint/policy_details.ts
+++ b/x-pack/test/security_solution_endpoint/apps/endpoint/policy_details.ts
@@ -110,9 +110,11 @@ export default function ({ getPageObjects, getService }: FtrProviderContext) {
               id: policyInfo.datasource.id,
               dataset: { namespace: 'default' },
               name: 'Protect East Coast',
-              package: {
-                name: 'endpoint',
-                version: policyInfo.packageInfo.version,
+              meta: {
+                package: {
+                  name: 'endpoint',
+                  version: policyInfo.packageInfo.version,
+                },
               },
               policy: {
                 linux: {