diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_config.tsx b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_config.tsx
index 356739af1ff9..0e8763cb2d4c 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_config.tsx
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_config.tsx
@@ -6,26 +6,38 @@
 import React, { useState, Fragment } from 'react';
 import { FormattedMessage } from '@kbn/i18n/react';
 import {
-  EuiFlexGrid,
   EuiFlexGroup,
   EuiFlexItem,
   EuiText,
+  EuiTextColor,
   EuiSpacer,
   EuiButtonEmpty,
   EuiTitle,
+  EuiIconTip,
 } from '@elastic/eui';
 import { DatasourceInput, RegistryVarsEntry } from '../../../../types';
-import { isAdvancedVar } from '../services';
+import { isAdvancedVar, DatasourceConfigValidationResults, validationHasErrors } from '../services';
 import { DatasourceInputVarField } from './datasource_input_var_field';
 
 export const DatasourceInputConfig: React.FunctionComponent<{
   packageInputVars?: RegistryVarsEntry[];
   datasourceInput: DatasourceInput;
   updateDatasourceInput: (updatedInput: Partial<DatasourceInput>) => void;
-}> = ({ packageInputVars, datasourceInput, updateDatasourceInput }) => {
+  inputVarsValidationResults: DatasourceConfigValidationResults;
+  forceShowErrors?: boolean;
+}> = ({
+  packageInputVars,
+  datasourceInput,
+  updateDatasourceInput,
+  inputVarsValidationResults,
+  forceShowErrors,
+}) => {
   // Showing advanced options toggle state
   const [isShowingAdvanced, setIsShowingAdvanced] = useState<boolean>(false);
 
+  // Errors state
+  const hasErrors = forceShowErrors && validationHasErrors(inputVarsValidationResults);
+
   const requiredVars: RegistryVarsEntry[] = [];
   const advancedVars: RegistryVarsEntry[] = [];
 
@@ -40,15 +52,36 @@ export const DatasourceInputConfig: React.FunctionComponent<{
   }
 
   return (
-    <EuiFlexGrid columns={2}>
-      <EuiFlexItem>
+    <EuiFlexGroup alignItems="flexStart">
+      <EuiFlexItem grow={1}>
         <EuiTitle size="s">
-          <h4>
-            <FormattedMessage
-              id="xpack.ingestManager.createDatasource.stepConfigure.inputSettingsTitle"
-              defaultMessage="Settings"
-            />
-          </h4>
+          <EuiFlexGroup alignItems="center" gutterSize="s">
+            <EuiFlexItem grow={false}>
+              <h4>
+                <EuiTextColor color={hasErrors ? 'danger' : 'default'}>
+                  <FormattedMessage
+                    id="xpack.ingestManager.createDatasource.stepConfigure.inputSettingsTitle"
+                    defaultMessage="Settings"
+                  />
+                </EuiTextColor>
+              </h4>
+            </EuiFlexItem>
+            {hasErrors ? (
+              <EuiFlexItem grow={false}>
+                <EuiIconTip
+                  content={
+                    <FormattedMessage
+                      id="xpack.ingestManager.createDatasource.stepConfigure.inputConfigErrorsTooltip"
+                      defaultMessage="Fix configuration errors"
+                    />
+                  }
+                  position="right"
+                  type="alert"
+                  iconProps={{ color: 'danger' }}
+                />
+              </EuiFlexItem>
+            ) : null}
+          </EuiFlexGroup>
         </EuiTitle>
         <EuiSpacer size="m" />
         <EuiText color="subdued" size="s">
@@ -60,7 +93,7 @@ export const DatasourceInputConfig: React.FunctionComponent<{
           </p>
         </EuiText>
       </EuiFlexItem>
-      <EuiFlexItem>
+      <EuiFlexItem grow={1}>
         <EuiFlexGroup direction="column" gutterSize="m">
           {requiredVars.map(varDef => {
             const { name: varName, type: varType } = varDef;
@@ -81,6 +114,8 @@ export const DatasourceInputConfig: React.FunctionComponent<{
                       },
                     });
                   }}
+                  errors={inputVarsValidationResults.config![varName]}
+                  forceShowErrors={forceShowErrors}
                 />
               </EuiFlexItem>
             );
@@ -123,6 +158,8 @@ export const DatasourceInputConfig: React.FunctionComponent<{
                               },
                             });
                           }}
+                          errors={inputVarsValidationResults.config![varName]}
+                          forceShowErrors={forceShowErrors}
                         />
                       </EuiFlexItem>
                     );
@@ -132,6 +169,6 @@ export const DatasourceInputConfig: React.FunctionComponent<{
           ) : null}
         </EuiFlexGroup>
       </EuiFlexItem>
-    </EuiFlexGrid>
+    </EuiFlexGroup>
   );
 };
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 74b08f48df12..6b0c68ccb7d3 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
@@ -17,8 +17,10 @@ import {
   EuiButtonIcon,
   EuiHorizontalRule,
   EuiSpacer,
+  EuiIconTip,
 } from '@elastic/eui';
 import { DatasourceInput, DatasourceInputStream, RegistryInput } from '../../../../types';
+import { DatasourceInputValidationResults, validationHasErrors } from '../services';
 import { DatasourceInputConfig } from './datasource_input_config';
 import { DatasourceInputStreamConfig } from './datasource_input_stream_config';
 
@@ -32,10 +34,21 @@ export const DatasourceInputPanel: React.FunctionComponent<{
   packageInput: RegistryInput;
   datasourceInput: DatasourceInput;
   updateDatasourceInput: (updatedInput: Partial<DatasourceInput>) => void;
-}> = ({ packageInput, datasourceInput, updateDatasourceInput }) => {
+  inputValidationResults: DatasourceInputValidationResults;
+  forceShowErrors?: boolean;
+}> = ({
+  packageInput,
+  datasourceInput,
+  updateDatasourceInput,
+  inputValidationResults,
+  forceShowErrors,
+}) => {
   // Showing streams toggle state
   const [isShowingStreams, setIsShowingStreams] = useState<boolean>(false);
 
+  // Errors state
+  const hasErrors = forceShowErrors && validationHasErrors(inputValidationResults);
+
   return (
     <EuiPanel>
       {/* Header / input-level toggle */}
@@ -43,9 +56,32 @@ export const DatasourceInputPanel: React.FunctionComponent<{
         <EuiFlexItem grow={false}>
           <EuiSwitch
             label={
-              <EuiText>
-                <h4>{packageInput.title || packageInput.type}</h4>
-              </EuiText>
+              <EuiFlexGroup alignItems="center" gutterSize="s">
+                <EuiFlexItem grow={false}>
+                  <EuiText>
+                    <h4>
+                      <EuiTextColor color={hasErrors ? 'danger' : 'default'}>
+                        {packageInput.title || packageInput.type}
+                      </EuiTextColor>
+                    </h4>
+                  </EuiText>
+                </EuiFlexItem>
+                {hasErrors ? (
+                  <EuiFlexItem grow={false}>
+                    <EuiIconTip
+                      content={
+                        <FormattedMessage
+                          id="xpack.ingestManager.createDatasource.stepConfigure.inputLevelErrorsTooltip"
+                          defaultMessage="Fix configuration errors"
+                        />
+                      }
+                      position="right"
+                      type="alert"
+                      iconProps={{ color: 'danger' }}
+                    />
+                  </EuiFlexItem>
+                ) : null}
+              </EuiFlexGroup>
             }
             checked={datasourceInput.enabled}
             onChange={e => {
@@ -122,6 +158,8 @@ export const DatasourceInputPanel: React.FunctionComponent<{
             packageInputVars={packageInput.vars}
             datasourceInput={datasourceInput}
             updateDatasourceInput={updateDatasourceInput}
+            inputVarsValidationResults={{ config: inputValidationResults.config }}
+            forceShowErrors={forceShowErrors}
           />
           <EuiHorizontalRule margin="m" />
         </Fragment>
@@ -165,6 +203,10 @@ export const DatasourceInputPanel: React.FunctionComponent<{
 
                     updateDatasourceInput(updatedInput);
                   }}
+                  inputStreamValidationResults={
+                    inputValidationResults.streams![datasourceInputStream.id]
+                  }
+                  forceShowErrors={forceShowErrors}
                 />
                 <EuiSpacer size="m" />
                 <EuiHorizontalRule margin="none" />
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 3bf5b2bb4c0f..43e8f5a2c060 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
@@ -7,26 +7,38 @@ import React, { useState, Fragment } from 'react';
 import ReactMarkdown from 'react-markdown';
 import { FormattedMessage } from '@kbn/i18n/react';
 import {
-  EuiFlexGrid,
   EuiFlexGroup,
   EuiFlexItem,
   EuiSwitch,
   EuiText,
   EuiSpacer,
   EuiButtonEmpty,
+  EuiTextColor,
+  EuiIconTip,
 } from '@elastic/eui';
 import { DatasourceInputStream, RegistryStream, RegistryVarsEntry } from '../../../../types';
-import { isAdvancedVar } from '../services';
+import { isAdvancedVar, DatasourceConfigValidationResults, validationHasErrors } from '../services';
 import { DatasourceInputVarField } from './datasource_input_var_field';
 
 export const DatasourceInputStreamConfig: React.FunctionComponent<{
   packageInputStream: RegistryStream;
   datasourceInputStream: DatasourceInputStream;
   updateDatasourceInputStream: (updatedStream: Partial<DatasourceInputStream>) => void;
-}> = ({ packageInputStream, datasourceInputStream, updateDatasourceInputStream }) => {
+  inputStreamValidationResults: DatasourceConfigValidationResults;
+  forceShowErrors?: boolean;
+}> = ({
+  packageInputStream,
+  datasourceInputStream,
+  updateDatasourceInputStream,
+  inputStreamValidationResults,
+  forceShowErrors,
+}) => {
   // Showing advanced options toggle state
   const [isShowingAdvanced, setIsShowingAdvanced] = useState<boolean>(false);
 
+  // Errors state
+  const hasErrors = forceShowErrors && validationHasErrors(inputStreamValidationResults);
+
   const requiredVars: RegistryVarsEntry[] = [];
   const advancedVars: RegistryVarsEntry[] = [];
 
@@ -41,10 +53,33 @@ export const DatasourceInputStreamConfig: React.FunctionComponent<{
   }
 
   return (
-    <EuiFlexGrid columns={2}>
-      <EuiFlexItem>
+    <EuiFlexGroup>
+      <EuiFlexItem grow={1}>
         <EuiSwitch
-          label={packageInputStream.title || packageInputStream.dataset}
+          label={
+            <EuiFlexGroup alignItems="center" gutterSize="s">
+              <EuiFlexItem grow={false}>
+                <EuiTextColor color={hasErrors ? 'danger' : 'default'}>
+                  {packageInputStream.title || packageInputStream.dataset}
+                </EuiTextColor>
+              </EuiFlexItem>
+              {hasErrors ? (
+                <EuiFlexItem grow={false}>
+                  <EuiIconTip
+                    content={
+                      <FormattedMessage
+                        id="xpack.ingestManager.createDatasource.stepConfigure.streamLevelErrorsTooltip"
+                        defaultMessage="Fix configuration errors"
+                      />
+                    }
+                    position="right"
+                    type="alert"
+                    iconProps={{ color: 'danger' }}
+                  />
+                </EuiFlexItem>
+              ) : null}
+            </EuiFlexGroup>
+          }
           checked={datasourceInputStream.enabled}
           onChange={e => {
             const enabled = e.target.checked;
@@ -62,7 +97,7 @@ export const DatasourceInputStreamConfig: React.FunctionComponent<{
           </Fragment>
         ) : null}
       </EuiFlexItem>
-      <EuiFlexItem>
+      <EuiFlexItem grow={1}>
         <EuiFlexGroup direction="column" gutterSize="m">
           {requiredVars.map(varDef => {
             const { name: varName, type: varType } = varDef;
@@ -83,6 +118,8 @@ export const DatasourceInputStreamConfig: React.FunctionComponent<{
                       },
                     });
                   }}
+                  errors={inputStreamValidationResults.config![varName]}
+                  forceShowErrors={forceShowErrors}
                 />
               </EuiFlexItem>
             );
@@ -125,6 +162,8 @@ export const DatasourceInputStreamConfig: React.FunctionComponent<{
                               },
                             });
                           }}
+                          errors={inputStreamValidationResults.config![varName]}
+                          forceShowErrors={forceShowErrors}
                         />
                       </EuiFlexItem>
                     );
@@ -134,6 +173,6 @@ export const DatasourceInputStreamConfig: React.FunctionComponent<{
           ) : null}
         </EuiFlexGroup>
       </EuiFlexItem>
-    </EuiFlexGrid>
+    </EuiFlexGroup>
   );
 };
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_var_field.tsx b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_var_field.tsx
index bcb99eed88ac..846a807f9240 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_var_field.tsx
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/datasource_input_var_field.tsx
@@ -3,7 +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 React from 'react';
+import React, { useState } from 'react';
 import ReactMarkdown from 'react-markdown';
 import { FormattedMessage } from '@kbn/i18n/react';
 import { EuiFormRow, EuiFieldText, EuiComboBox, EuiText, EuiCodeEditor } from '@elastic/eui';
@@ -16,12 +16,20 @@ export const DatasourceInputVarField: React.FunctionComponent<{
   varDef: RegistryVarsEntry;
   value: any;
   onChange: (newValue: any) => void;
-}> = ({ varDef, value, onChange }) => {
+  errors?: string[] | null;
+  forceShowErrors?: boolean;
+}> = ({ varDef, value, onChange, errors: varErrors, forceShowErrors }) => {
+  const [isDirty, setIsDirty] = useState<boolean>(false);
+  const { multi, required, type, title, name, description } = varDef;
+  const isInvalid = (isDirty || forceShowErrors) && !!varErrors;
+  const errors = isInvalid ? varErrors : null;
+
   const renderField = () => {
-    if (varDef.multi) {
+    if (multi) {
       return (
         <EuiComboBox
           noSuggestions
+          isInvalid={isInvalid}
           selectedOptions={value.map((val: string) => ({ label: val }))}
           onCreateOption={(newVal: any) => {
             onChange([...value, newVal]);
@@ -29,10 +37,11 @@ export const DatasourceInputVarField: React.FunctionComponent<{
           onChange={(newVals: any[]) => {
             onChange(newVals.map(val => val.label));
           }}
+          onBlur={() => setIsDirty(true)}
         />
       );
     }
-    if (varDef.type === 'yaml') {
+    if (type === 'yaml') {
       return (
         <EuiCodeEditor
           width="100%"
@@ -46,22 +55,27 @@ export const DatasourceInputVarField: React.FunctionComponent<{
           }}
           value={value}
           onChange={newVal => onChange(newVal)}
+          onBlur={() => setIsDirty(true)}
         />
       );
     }
     return (
       <EuiFieldText
+        isInvalid={isInvalid}
         value={value === undefined ? '' : value}
         onChange={e => onChange(e.target.value)}
+        onBlur={() => setIsDirty(true)}
       />
     );
   };
 
   return (
     <EuiFormRow
-      label={varDef.title || varDef.name}
+      isInvalid={isInvalid}
+      error={errors}
+      label={title || name}
       labelAppend={
-        !varDef.required ? (
+        !required ? (
           <EuiText size="xs" color="subdued">
             <FormattedMessage
               id="xpack.ingestManager.createDatasource.stepConfigure.inputVarFieldOptionalLabel"
@@ -70,7 +84,7 @@ export const DatasourceInputVarField: React.FunctionComponent<{
           </EuiText>
         ) : null
       }
-      helpText={<ReactMarkdown source={varDef.description} />}
+      helpText={<ReactMarkdown source={description} />}
     >
       {renderField()}
     </EuiFormRow>
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/index.ts b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/index.ts
index e5f18e1449d1..3bfca7566891 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/index.ts
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/components/index.ts
@@ -5,3 +5,4 @@
  */
 export { CreateDatasourcePageLayout } from './layout';
 export { DatasourceInputPanel } from './datasource_input_panel';
+export { DatasourceInputVarField } from './datasource_input_var_field';
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/index.tsx b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/index.tsx
index 23d0f3317a66..7815ab9cd1d6 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/index.tsx
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/index.tsx
@@ -21,6 +21,7 @@ import { useLinks as useEPMLinks } from '../../epm/hooks';
 import { CreateDatasourcePageLayout } from './components';
 import { CreateDatasourceFrom, CreateDatasourceStep } from './types';
 import { CREATE_DATASOURCE_STEP_PATHS } from './constants';
+import { DatasourceValidationResults, validateDatasource } from './services';
 import { StepSelectPackage } from './step_select_package';
 import { StepSelectConfig } from './step_select_config';
 import { StepConfigureDatasource } from './step_configure_datasource';
@@ -51,6 +52,9 @@ export const CreateDatasourcePage: React.FunctionComponent = () => {
     inputs: [],
   });
 
+  // Datasource validation state
+  const [validationResults, setValidationResults] = useState<DatasourceValidationResults>();
+
   // Update package info method
   const updatePackageInfo = (updatedPackageInfo: PackageInfo | undefined) => {
     if (updatedPackageInfo) {
@@ -84,9 +88,18 @@ export const CreateDatasourcePage: React.FunctionComponent = () => {
       ...updatedFields,
     };
     setDatasource(newDatasource);
-
     // eslint-disable-next-line no-console
     console.debug('Datasource updated', newDatasource);
+    updateDatasourceValidation(newDatasource);
+  };
+
+  const updateDatasourceValidation = (newDatasource?: NewDatasource) => {
+    if (packageInfo) {
+      const newValidationResult = validateDatasource(newDatasource || datasource, packageInfo);
+      setValidationResults(newValidationResult);
+      // eslint-disable-next-line no-console
+      console.debug('Datasource validation results', newValidationResult);
+    }
   };
 
   // Cancel url
@@ -202,6 +215,7 @@ export const CreateDatasourcePage: React.FunctionComponent = () => {
                 packageInfo={packageInfo}
                 datasource={datasource}
                 updateDatasource={updateDatasource}
+                validationResults={validationResults!}
                 backLink={
                   <EuiButtonEmpty href={firstStepUrl} iconType="arrowLeft" iconSide="left">
                     {from === 'config' ? (
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/index.ts b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/index.ts
index 44e5bfa41cb9..d99f0712db3c 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/index.ts
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/index.ts
@@ -4,3 +4,10 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 export { isAdvancedVar } from './is_advanced_var';
+export {
+  DatasourceValidationResults,
+  DatasourceConfigValidationResults,
+  DatasourceInputValidationResults,
+  validateDatasource,
+  validationHasErrors,
+} from './validate_datasource';
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
new file mode 100644
index 000000000000..a45fabeb5ed6
--- /dev/null
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.test.ts
@@ -0,0 +1,504 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+import {
+  PackageInfo,
+  InstallationStatus,
+  NewDatasource,
+  RegistryDatasource,
+} from '../../../../types';
+import { validateDatasource, validationHasErrors } from './validate_datasource';
+
+describe('Ingest Manager - validateDatasource()', () => {
+  const mockPackage = ({
+    name: 'mock-package',
+    title: 'Mock package',
+    version: '0.0.0',
+    description: 'description',
+    type: 'mock',
+    categories: [],
+    requirement: { kibana: { versions: '' }, elasticsearch: { versions: '' } },
+    format_version: '',
+    download: '',
+    path: '',
+    assets: {
+      kibana: {
+        dashboard: [],
+        visualization: [],
+        search: [],
+        'index-pattern': [],
+      },
+    },
+    status: InstallationStatus.notInstalled,
+    datasources: [
+      {
+        name: 'datasource1',
+        title: 'Datasource 1',
+        description: 'test datasource',
+        inputs: [
+          {
+            type: 'foo',
+            title: 'Foo',
+            vars: [
+              { default: 'foo-input-var-value', name: 'foo-input-var-name', type: 'text' },
+              {
+                default: 'foo-input2-var-value',
+                name: 'foo-input2-var-name',
+                required: true,
+                type: 'text',
+              },
+              { 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',
+            title: 'Bar',
+            vars: [
+              {
+                default: ['value1', 'value2'],
+                name: 'bar-input-var-name',
+                type: 'text',
+                multi: true,
+              },
+              { 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 },
+            ],
+          },
+        ],
+      },
+    ],
+  } as unknown) as PackageInfo;
+
+  const validDatasource: NewDatasource = {
+    name: 'datasource1-1',
+    config_id: 'test-config',
+    enabled: true,
+    output_id: 'test-output',
+    inputs: [
+      {
+        type: 'foo',
+        enabled: true,
+        config: {
+          'foo-input-var-name': { value: 'foo-input-var-value', type: 'text' },
+          'foo-input2-var-name': { value: 'foo-input2-var-value', type: 'text' },
+          'foo-input3-var-name': { value: ['test'], type: 'text' },
+        },
+        streams: [
+          {
+            id: 'foo-foo',
+            dataset: 'foo',
+            enabled: true,
+            config: { 'var-name': { value: 'test_yaml: value', type: 'yaml' } },
+          },
+        ],
+      },
+      {
+        type: 'bar',
+        enabled: true,
+        config: {
+          'bar-input-var-name': { value: ['value1', 'value2'], type: 'text' },
+          'bar-input2-var-name': { value: 'test', type: 'text' },
+        },
+        streams: [
+          {
+            id: 'bar-bar',
+            dataset: 'bar',
+            enabled: true,
+            config: { 'var-name': { value: 'test_yaml: value', type: 'yaml' } },
+          },
+          {
+            id: 'bar-bar2',
+            dataset: 'bar2',
+            enabled: true,
+            config: { 'var-name': { value: undefined, type: 'text' } },
+          },
+        ],
+      },
+      {
+        type: 'with-no-config-or-streams',
+        enabled: true,
+        streams: [],
+      },
+      {
+        type: 'with-disabled-streams',
+        enabled: true,
+        streams: [
+          {
+            id: 'with-disabled-streams-disabled',
+            dataset: 'disabled',
+            enabled: false,
+            config: { 'var-name': { value: undefined, type: 'text' } },
+          },
+          {
+            id: 'with-disabled-streams-disabled2',
+            dataset: 'disabled2',
+            enabled: false,
+          },
+        ],
+      },
+    ],
+  };
+
+  const invalidDatasource: NewDatasource = {
+    ...validDatasource,
+    name: '',
+    inputs: [
+      {
+        type: 'foo',
+        enabled: true,
+        config: {
+          'foo-input-var-name': { value: undefined, type: 'text' },
+          'foo-input2-var-name': { value: '', type: 'text' },
+          'foo-input3-var-name': { value: [], type: 'text' },
+        },
+        streams: [
+          {
+            id: 'foo-foo',
+            dataset: 'foo',
+            enabled: true,
+            config: { 'var-name': { value: 'invalidyaml: test\n foo bar:', type: 'yaml' } },
+          },
+        ],
+      },
+      {
+        type: 'bar',
+        enabled: true,
+        config: {
+          'bar-input-var-name': { value: 'invalid value for multi', type: 'text' },
+          'bar-input2-var-name': { value: undefined, type: 'text' },
+        },
+        streams: [
+          {
+            id: 'bar-bar',
+            dataset: 'bar',
+            enabled: true,
+            config: { 'var-name': { value: '    \n\n', type: 'yaml' } },
+          },
+          {
+            id: 'bar-bar2',
+            dataset: 'bar2',
+            enabled: true,
+            config: { 'var-name': { value: undefined, type: 'text' } },
+          },
+        ],
+      },
+      {
+        type: 'with-no-config-or-streams',
+        enabled: true,
+        streams: [],
+      },
+      {
+        type: 'with-disabled-streams',
+        enabled: true,
+        streams: [
+          {
+            id: 'with-disabled-streams-disabled',
+            dataset: 'disabled',
+            enabled: false,
+            config: {
+              'var-name': {
+                value: 'invalid value but not checked due to not enabled',
+                type: 'text',
+              },
+            },
+          },
+          {
+            id: 'with-disabled-streams-disabled2',
+            dataset: 'disabled2',
+            enabled: false,
+          },
+        ],
+      },
+    ],
+  };
+
+  const noErrorsValidationResults = {
+    name: null,
+    description: null,
+    inputs: {
+      foo: {
+        config: {
+          'foo-input-var-name': null,
+          'foo-input2-var-name': null,
+          'foo-input3-var-name': null,
+        },
+        streams: { 'foo-foo': { config: { 'var-name': null } } },
+      },
+      bar: {
+        config: { 'bar-input-var-name': null, 'bar-input2-var-name': null },
+        streams: {
+          'bar-bar': { config: { 'var-name': null } },
+          'bar-bar2': { config: { 'var-name': null } },
+        },
+      },
+      'with-disabled-streams': {
+        streams: { 'with-disabled-streams-disabled': { config: { 'var-name': null } } },
+      },
+    },
+  };
+
+  it('returns no errors for valid datasource configuration', () => {
+    expect(validateDatasource(validDatasource, mockPackage)).toEqual(noErrorsValidationResults);
+  });
+
+  it('returns errors for invalid datasource configuration', () => {
+    expect(validateDatasource(invalidDatasource, mockPackage)).toEqual({
+      name: ['Name is required'],
+      description: null,
+      inputs: {
+        foo: {
+          config: {
+            'foo-input-var-name': null,
+            'foo-input2-var-name': ['foo-input2-var-name is required'],
+            'foo-input3-var-name': ['foo-input3-var-name is required'],
+          },
+          streams: { 'foo-foo': { config: { 'var-name': ['Invalid YAML format'] } } },
+        },
+        bar: {
+          config: {
+            'bar-input-var-name': ['Invalid format'],
+            'bar-input2-var-name': ['bar-input2-var-name is required'],
+          },
+          streams: {
+            'bar-bar': { config: { 'var-name': ['var-name is required'] } },
+            'bar-bar2': { config: { 'var-name': null } },
+          },
+        },
+        'with-disabled-streams': {
+          streams: { 'with-disabled-streams-disabled': { config: { 'var-name': null } } },
+        },
+      },
+    });
+  });
+
+  it('returns no errors for disabled inputs', () => {
+    const disabledInputs = invalidDatasource.inputs.map(input => ({ ...input, enabled: false }));
+    expect(validateDatasource({ ...validDatasource, inputs: disabledInputs }, mockPackage)).toEqual(
+      noErrorsValidationResults
+    );
+  });
+
+  it('returns only datasource and input-level errors for disabled streams', () => {
+    const inputsWithDisabledStreams = invalidDatasource.inputs.map(input =>
+      input.streams
+        ? {
+            ...input,
+            streams: input.streams.map(stream => ({ ...stream, enabled: false })),
+          }
+        : input
+    );
+    expect(
+      validateDatasource({ ...invalidDatasource, inputs: inputsWithDisabledStreams }, mockPackage)
+    ).toEqual({
+      name: ['Name is required'],
+      description: null,
+      inputs: {
+        foo: {
+          config: {
+            'foo-input-var-name': null,
+            'foo-input2-var-name': ['foo-input2-var-name is required'],
+            'foo-input3-var-name': ['foo-input3-var-name is required'],
+          },
+          streams: { 'foo-foo': { config: { 'var-name': null } } },
+        },
+        bar: {
+          config: {
+            'bar-input-var-name': ['Invalid format'],
+            'bar-input2-var-name': ['bar-input2-var-name is required'],
+          },
+          streams: {
+            'bar-bar': { config: { 'var-name': null } },
+            'bar-bar2': { config: { 'var-name': null } },
+          },
+        },
+        'with-disabled-streams': {
+          streams: { 'with-disabled-streams-disabled': { config: { 'var-name': null } } },
+        },
+      },
+    });
+  });
+
+  it('returns no errors for packages with no datasources', () => {
+    expect(
+      validateDatasource(validDatasource, {
+        ...mockPackage,
+        datasources: undefined,
+      })
+    ).toEqual({
+      name: null,
+      description: null,
+      inputs: null,
+    });
+    expect(
+      validateDatasource(validDatasource, {
+        ...mockPackage,
+        datasources: [],
+      })
+    ).toEqual({
+      name: null,
+      description: null,
+      inputs: null,
+    });
+  });
+
+  it('returns no errors for packages with no inputs', () => {
+    expect(
+      validateDatasource(validDatasource, {
+        ...mockPackage,
+        datasources: [{} as RegistryDatasource],
+      })
+    ).toEqual({
+      name: null,
+      description: null,
+      inputs: null,
+    });
+    expect(
+      validateDatasource(validDatasource, {
+        ...mockPackage,
+        datasources: [({ inputs: [] } as unknown) as RegistryDatasource],
+      })
+    ).toEqual({
+      name: null,
+      description: null,
+      inputs: null,
+    });
+  });
+});
+
+describe('Ingest Manager - validationHasErrors()', () => {
+  it('returns true for stream validation results with errors', () => {
+    expect(
+      validationHasErrors({
+        config: { foo: ['foo error'], bar: null },
+      })
+    ).toBe(true);
+  });
+
+  it('returns false for stream validation results with no errors', () => {
+    expect(
+      validationHasErrors({
+        config: { foo: null, bar: null },
+      })
+    ).toBe(false);
+  });
+
+  it('returns true for input validation results with errors', () => {
+    expect(
+      validationHasErrors({
+        config: { foo: ['foo error'], bar: null },
+        streams: { stream1: { config: { foo: null, bar: null } } },
+      })
+    ).toBe(true);
+    expect(
+      validationHasErrors({
+        config: { foo: null, bar: null },
+        streams: { stream1: { config: { foo: ['foo error'], bar: null } } },
+      })
+    ).toBe(true);
+  });
+
+  it('returns false for input validation results with no errors', () => {
+    expect(
+      validationHasErrors({
+        config: { foo: null, bar: null },
+        streams: { stream1: { config: { foo: null, bar: null } } },
+      })
+    ).toBe(false);
+  });
+
+  it('returns true for datasource validation results with errors', () => {
+    expect(
+      validationHasErrors({
+        name: ['name error'],
+        description: null,
+        inputs: {
+          input1: {
+            config: { foo: null, bar: null },
+            streams: { stream1: { config: { foo: null, bar: null } } },
+          },
+        },
+      })
+    ).toBe(true);
+    expect(
+      validationHasErrors({
+        name: null,
+        description: null,
+        inputs: {
+          input1: {
+            config: { foo: ['foo error'], bar: null },
+            streams: { stream1: { config: { foo: null, bar: null } } },
+          },
+        },
+      })
+    ).toBe(true);
+    expect(
+      validationHasErrors({
+        name: null,
+        description: null,
+        inputs: {
+          input1: {
+            config: { foo: null, bar: null },
+            streams: { stream1: { config: { foo: ['foo error'], bar: null } } },
+          },
+        },
+      })
+    ).toBe(true);
+  });
+
+  it('returns false for datasource validation results with no errors', () => {
+    expect(
+      validationHasErrors({
+        name: null,
+        description: null,
+        inputs: {
+          input1: {
+            config: { foo: null, bar: null },
+            streams: { stream1: { config: { foo: null, bar: null } } },
+          },
+        },
+      })
+    ).toBe(false);
+  });
+});
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
new file mode 100644
index 000000000000..518e2bfc1af0
--- /dev/null
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/sections/agent_config/create_datasource_page/services/validate_datasource.ts
@@ -0,0 +1,232 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+import { i18n } from '@kbn/i18n';
+import { safeLoad } from 'js-yaml';
+import { getFlattenedObject } from '../../../../services';
+import {
+  NewDatasource,
+  DatasourceInput,
+  DatasourceInputStream,
+  DatasourceConfigRecordEntry,
+  PackageInfo,
+  RegistryInput,
+  RegistryVarsEntry,
+} from '../../../../types';
+
+type Errors = string[] | null;
+
+type ValidationEntry = Record<string, Errors>;
+
+export interface DatasourceConfigValidationResults {
+  config?: ValidationEntry;
+}
+
+export type DatasourceInputValidationResults = DatasourceConfigValidationResults & {
+  streams?: Record<DatasourceInputStream['id'], DatasourceConfigValidationResults>;
+};
+
+export interface DatasourceValidationResults {
+  name: Errors;
+  description: Errors;
+  inputs: Record<DatasourceInput['type'], DatasourceInputValidationResults> | null;
+}
+
+/*
+ * Returns validation information for a given datasource configuration and package info
+ * Note: this method assumes that `datasource` is correctly structured for the given package
+ */
+export const validateDatasource = (
+  datasource: NewDatasource,
+  packageInfo: PackageInfo
+): DatasourceValidationResults => {
+  const validationResults: DatasourceValidationResults = {
+    name: null,
+    description: null,
+    inputs: {},
+  };
+
+  if (!datasource.name.trim()) {
+    validationResults.name = [
+      i18n.translate('xpack.ingestManager.datasourceValidation.nameRequiredErrorMessage', {
+        defaultMessage: 'Name is required',
+      }),
+    ];
+  }
+
+  if (
+    !packageInfo.datasources ||
+    packageInfo.datasources.length === 0 ||
+    !packageInfo.datasources[0] ||
+    !packageInfo.datasources[0].inputs ||
+    packageInfo.datasources[0].inputs.length === 0
+  ) {
+    validationResults.inputs = null;
+    return validationResults;
+  }
+
+  const registryInputsByType: Record<
+    string,
+    RegistryInput
+  > = packageInfo.datasources[0].inputs.reduce((inputs, registryInput) => {
+    inputs[registryInput.type] = registryInput;
+    return inputs;
+  }, {} as Record<string, RegistryInput>);
+
+  // Validate each datasource input with either its own config fields or streams
+  datasource.inputs.forEach(input => {
+    if (!input.config && !input.streams) {
+      return;
+    }
+
+    const inputValidationResults: DatasourceInputValidationResults = {
+      config: undefined,
+      streams: {},
+    };
+
+    const inputVarsByName = (registryInputsByType[input.type].vars || []).reduce(
+      (vars, registryVar) => {
+        vars[registryVar.name] = registryVar;
+        return vars;
+      },
+      {} as Record<string, RegistryVarsEntry>
+    );
+
+    // Validate input-level config fields
+    const inputConfigs = Object.entries(input.config || {});
+    if (inputConfigs.length) {
+      inputValidationResults.config = inputConfigs.reduce((results, [name, configEntry]) => {
+        results[name] = input.enabled
+          ? validateDatasourceConfig(configEntry, inputVarsByName[name])
+          : null;
+        return results;
+      }, {} as ValidationEntry);
+    } else {
+      delete inputValidationResults.config;
+    }
+
+    // Validate each input stream with config fields
+    if (input.streams.length) {
+      input.streams.forEach(stream => {
+        if (!stream.config) {
+          return;
+        }
+
+        const streamValidationResults: DatasourceConfigValidationResults = {
+          config: undefined,
+        };
+
+        const streamVarsByName = (
+          (
+            registryInputsByType[input.type].streams.find(
+              registryStream => registryStream.dataset === stream.dataset
+            ) || {}
+          ).vars || []
+        ).reduce((vars, registryVar) => {
+          vars[registryVar.name] = registryVar;
+          return vars;
+        }, {} as Record<string, RegistryVarsEntry>);
+
+        // Validate stream-level config fields
+        streamValidationResults.config = Object.entries(stream.config).reduce(
+          (results, [name, configEntry]) => {
+            results[name] =
+              input.enabled && stream.enabled
+                ? validateDatasourceConfig(configEntry, streamVarsByName[name])
+                : null;
+            return results;
+          },
+          {} as ValidationEntry
+        );
+
+        inputValidationResults.streams![stream.id] = streamValidationResults;
+      });
+    } else {
+      delete inputValidationResults.streams;
+    }
+
+    if (inputValidationResults.config || inputValidationResults.streams) {
+      validationResults.inputs![input.type] = inputValidationResults;
+    }
+  });
+
+  if (Object.entries(validationResults.inputs!).length === 0) {
+    validationResults.inputs = null;
+  }
+  return validationResults;
+};
+
+const validateDatasourceConfig = (
+  configEntry: DatasourceConfigRecordEntry,
+  varDef: RegistryVarsEntry
+): string[] | null => {
+  const errors = [];
+  const { value } = configEntry;
+  let parsedValue: any = value;
+
+  if (typeof value === 'string') {
+    parsedValue = value.trim();
+  }
+
+  if (varDef.required) {
+    if (parsedValue === undefined || (typeof parsedValue === 'string' && !parsedValue)) {
+      errors.push(
+        i18n.translate('xpack.ingestManager.datasourceValidation.requiredErrorMessage', {
+          defaultMessage: '{fieldName} is required',
+          values: {
+            fieldName: varDef.title || varDef.name,
+          },
+        })
+      );
+    }
+  }
+
+  if (varDef.type === 'yaml') {
+    try {
+      parsedValue = safeLoad(value);
+    } catch (e) {
+      errors.push(
+        i18n.translate('xpack.ingestManager.datasourceValidation.invalidYamlFormatErrorMessage', {
+          defaultMessage: 'Invalid YAML format',
+        })
+      );
+    }
+  }
+
+  if (varDef.multi) {
+    if (parsedValue && !Array.isArray(parsedValue)) {
+      errors.push(
+        i18n.translate('xpack.ingestManager.datasourceValidation.invalidArrayErrorMessage', {
+          defaultMessage: 'Invalid format',
+        })
+      );
+    }
+    if (
+      varDef.required &&
+      (!parsedValue || (Array.isArray(parsedValue) && parsedValue.length === 0))
+    ) {
+      errors.push(
+        i18n.translate('xpack.ingestManager.datasourceValidation.requiredErrorMessage', {
+          defaultMessage: '{fieldName} is required',
+          values: {
+            fieldName: varDef.title || varDef.name,
+          },
+        })
+      );
+    }
+  }
+
+  return errors.length ? errors : null;
+};
+
+export const validationHasErrors = (
+  validationResults:
+    | DatasourceValidationResults
+    | DatasourceInputValidationResults
+    | DatasourceConfigValidationResults
+) => {
+  const flattenedValidation = getFlattenedObject(validationResults);
+  return !!Object.entries(flattenedValidation).find(([, value]) => !!value);
+};
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 b45beef4a8b5..105d6c66a570 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
@@ -9,17 +9,16 @@ import { FormattedMessage } from '@kbn/i18n/react';
 import {
   EuiSteps,
   EuiPanel,
-  EuiFlexGrid,
   EuiFlexGroup,
   EuiFlexItem,
   EuiFormRow,
-  EuiFieldText,
   EuiButtonEmpty,
   EuiSpacer,
   EuiEmptyPrompt,
   EuiText,
   EuiButton,
   EuiComboBox,
+  EuiCallOut,
 } from '@elastic/eui';
 import {
   AgentConfig,
@@ -28,21 +27,37 @@ import {
   NewDatasource,
   DatasourceInput,
 } from '../../../types';
+import { Loading } from '../../../components';
 import { packageToConfigDatasourceInputs } from '../../../services';
-import { DatasourceInputPanel } from './components';
+import { DatasourceValidationResults, validationHasErrors } from './services';
+import { DatasourceInputPanel, DatasourceInputVarField } from './components';
 
 export const StepConfigureDatasource: React.FunctionComponent<{
   agentConfig: AgentConfig;
   packageInfo: PackageInfo;
   datasource: NewDatasource;
   updateDatasource: (fields: Partial<NewDatasource>) => void;
+  validationResults: DatasourceValidationResults;
   backLink: JSX.Element;
   cancelUrl: string;
   onNext: () => void;
-}> = ({ agentConfig, packageInfo, datasource, updateDatasource, backLink, cancelUrl, onNext }) => {
+}> = ({
+  agentConfig,
+  packageInfo,
+  datasource,
+  updateDatasource,
+  validationResults,
+  backLink,
+  cancelUrl,
+  onNext,
+}) => {
   // Form show/hide states
   const [isShowingAdvancedDefine, setIsShowingAdvancedDefine] = useState<boolean>(false);
 
+  // Form submit state
+  const [submitAttempted, setSubmitAttempted] = useState<boolean>(false);
+  const hasErrors = validationResults ? validationHasErrors(validationResults) : false;
+
   // Update datasource's package and config info
   useEffect(() => {
     const dsPackage = datasource.package;
@@ -81,56 +96,56 @@ export const StepConfigureDatasource: React.FunctionComponent<{
   }, [datasource.package, datasource.config_id, agentConfig, packageInfo, updateDatasource]);
 
   // Step A, define datasource
-  const DefineDatasource = (
+  const renderDefineDatasource = () => (
     <EuiPanel>
-      <EuiFlexGrid columns={2}>
-        <EuiFlexItem>
-          <EuiFormRow
-            label={
-              <FormattedMessage
-                id="xpack.ingestManager.createDatasource.stepConfigure.datasourceNameInputLabel"
-                defaultMessage="Data source name"
-              />
-            }
-          >
-            <EuiFieldText
-              value={datasource.name}
-              onChange={e =>
-                updateDatasource({
-                  name: e.target.value,
-                })
-              }
-            />
-          </EuiFormRow>
+      <EuiFlexGroup>
+        <EuiFlexItem grow={1}>
+          <DatasourceInputVarField
+            varDef={{
+              name: 'name',
+              title: i18n.translate(
+                'xpack.ingestManager.createDatasource.stepConfigure.datasourceNameInputLabel',
+                {
+                  defaultMessage: 'Data source name',
+                }
+              ),
+              type: 'text',
+              required: true,
+            }}
+            value={datasource.name}
+            onChange={(newValue: any) => {
+              updateDatasource({
+                name: newValue,
+              });
+            }}
+            errors={validationResults!.name}
+            forceShowErrors={submitAttempted}
+          />
         </EuiFlexItem>
-        <EuiFlexItem>
-          <EuiFormRow
-            label={
-              <FormattedMessage
-                id="xpack.ingestManager.createDatasource.stepConfigure.datasourceDescriptionInputLabel"
-                defaultMessage="Description"
-              />
-            }
-            labelAppend={
-              <EuiText size="xs" color="subdued">
-                <FormattedMessage
-                  id="xpack.ingestManager.createDatasource.stepConfigure.inputVarFieldOptionalLabel"
-                  defaultMessage="Optional"
-                />
-              </EuiText>
-            }
-          >
-            <EuiFieldText
-              value={datasource.description}
-              onChange={e =>
-                updateDatasource({
-                  description: e.target.value,
-                })
-              }
-            />
-          </EuiFormRow>
+        <EuiFlexItem grow={1}>
+          <DatasourceInputVarField
+            varDef={{
+              name: 'description',
+              title: i18n.translate(
+                'xpack.ingestManager.createDatasource.stepConfigure.datasourceDescriptionInputLabel',
+                {
+                  defaultMessage: 'Description',
+                }
+              ),
+              type: 'text',
+              required: false,
+            }}
+            value={datasource.description}
+            onChange={(newValue: any) => {
+              updateDatasource({
+                description: newValue,
+              });
+            }}
+            errors={validationResults!.description}
+            forceShowErrors={submitAttempted}
+          />
         </EuiFlexItem>
-      </EuiFlexGrid>
+      </EuiFlexGroup>
       <EuiSpacer size="m" />
       <EuiButtonEmpty
         flush="left"
@@ -147,8 +162,8 @@ export const StepConfigureDatasource: React.FunctionComponent<{
       {isShowingAdvancedDefine ? (
         <Fragment>
           <EuiSpacer size="m" />
-          <EuiFlexGrid columns={2}>
-            <EuiFlexItem>
+          <EuiFlexGroup>
+            <EuiFlexItem grow={1}>
               <EuiFormRow
                 label={
                   <FormattedMessage
@@ -174,7 +189,8 @@ export const StepConfigureDatasource: React.FunctionComponent<{
                 />
               </EuiFormRow>
             </EuiFlexItem>
-          </EuiFlexGrid>
+            <EuiFlexItem grow={1} />
+          </EuiFlexGroup>
         </Fragment>
       ) : null}
     </EuiPanel>
@@ -182,7 +198,7 @@ export const StepConfigureDatasource: React.FunctionComponent<{
 
   // Step B, configure inputs (and their streams)
   // Assume packages only export one datasource for now
-  const ConfigureInputs =
+  const renderConfigureInputs = () =>
     packageInfo.datasources &&
     packageInfo.datasources[0] &&
     packageInfo.datasources[0].inputs &&
@@ -208,6 +224,8 @@ export const StepConfigureDatasource: React.FunctionComponent<{
                     inputs: newInputs,
                   });
                 }}
+                inputValidationResults={validationResults!.inputs![datasourceInput.type]}
+                forceShowErrors={submitAttempted}
               />
             </EuiFlexItem>
           ) : null;
@@ -232,7 +250,7 @@ export const StepConfigureDatasource: React.FunctionComponent<{
       </EuiPanel>
     );
 
-  return (
+  return validationResults ? (
     <EuiFlexGroup direction="column" gutterSize="none">
       <EuiFlexItem>
         <EuiFlexGroup direction="column" gutterSize="none">
@@ -251,7 +269,7 @@ export const StepConfigureDatasource: React.FunctionComponent<{
                       defaultMessage: 'Define your datasource',
                     }
                   ),
-                  children: DefineDatasource,
+                  children: renderDefineDatasource(),
                 },
                 {
                   title: i18n.translate(
@@ -260,13 +278,34 @@ export const StepConfigureDatasource: React.FunctionComponent<{
                       defaultMessage: 'Choose the data you want to collect',
                     }
                   ),
-                  children: ConfigureInputs,
+                  children: renderConfigureInputs(),
                 },
               ]}
             />
           </EuiFlexItem>
         </EuiFlexGroup>
       </EuiFlexItem>
+      {hasErrors && submitAttempted ? (
+        <EuiFlexItem>
+          <EuiCallOut
+            title={i18n.translate(
+              'xpack.ingestManager.createDatasource.stepConfigure.validationErrorTitle',
+              {
+                defaultMessage: 'Your data source configuration has errors',
+              }
+            )}
+            color="danger"
+          >
+            <p>
+              <FormattedMessage
+                id="xpack.ingestManager.createDatasource.stepConfigure.validationErrorText"
+                defaultMessage="Please fix the above errors before continuing"
+              />
+            </p>
+          </EuiCallOut>
+          <EuiSpacer size="m" />
+        </EuiFlexItem>
+      ) : null}
       <EuiFlexItem>
         <EuiFlexGroup justifyContent="flexEnd">
           <EuiFlexItem grow={false}>
@@ -278,7 +317,17 @@ export const StepConfigureDatasource: React.FunctionComponent<{
             </EuiButtonEmpty>
           </EuiFlexItem>
           <EuiFlexItem grow={false}>
-            <EuiButton fill iconType="arrowRight" iconSide="right" onClick={() => onNext()}>
+            <EuiButton
+              fill
+              iconType="arrowRight"
+              iconSide="right"
+              onClick={() => {
+                setSubmitAttempted(true);
+                if (!hasErrors) {
+                  onNext();
+                }
+              }}
+            >
               <FormattedMessage
                 id="xpack.ingestManager.createDatasource.continueButtonText"
                 defaultMessage="Continue"
@@ -288,5 +337,7 @@ export const StepConfigureDatasource: React.FunctionComponent<{
         </EuiFlexGroup>
       </EuiFlexItem>
     </EuiFlexGroup>
+  ) : (
+    <Loading />
   );
 };
diff --git a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/services/index.ts b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/services/index.ts
index 0aa08602e4d4..5ebd1300baf6 100644
--- a/x-pack/plugins/ingest_manager/public/applications/ingest_manager/services/index.ts
+++ b/x-pack/plugins/ingest_manager/public/applications/ingest_manager/services/index.ts
@@ -4,6 +4,8 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+export { getFlattenedObject } from '../../../../../../../src/core/utils';
+
 export {
   agentConfigRouteService,
   datasourceRouteService,
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 333a9b049fa8..32615278b67d 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
@@ -16,6 +16,7 @@ export {
   NewDatasource,
   DatasourceInput,
   DatasourceInputStream,
+  DatasourceConfigRecordEntry,
   // API schemas - Agent Config
   GetAgentConfigsResponse,
   GetAgentConfigsResponseItem,
@@ -56,6 +57,7 @@ export {
   RegistryVarsEntry,
   RegistryInput,
   RegistryStream,
+  RegistryDatasource,
   PackageList,
   PackageListItem,
   PackagesGroupedByStatus,
@@ -70,4 +72,5 @@ export {
   DeletePackageResponse,
   DetailViewPanelName,
   InstallStatus,
+  InstallationStatus,
 } from '../../../../common';