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 356739af1ff9a..7694c35c35ad8 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
@@ -15,14 +15,22 @@ import {
   EuiTitle,
 } from '@elastic/eui';
 import { DatasourceInput, RegistryVarsEntry } from '../../../../types';
-import { isAdvancedVar } from '../services';
+import { isAdvancedVar, DatasourceInputValidationResults } 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: DatasourceInputValidationResults;
+  forceShowErrors?: boolean;
+}> = ({
+  packageInputVars,
+  datasourceInput,
+  updateDatasourceInput,
+  inputVarsValidationResults,
+  forceShowErrors,
+}) => {
   // Showing advanced options toggle state
   const [isShowingAdvanced, setIsShowingAdvanced] = useState<boolean>(false);
 
@@ -81,6 +89,8 @@ export const DatasourceInputConfig: React.FunctionComponent<{
                       },
                     });
                   }}
+                  errors={inputVarsValidationResults.config![varName]}
+                  forceShowErrors={forceShowErrors}
                 />
               </EuiFlexItem>
             );
@@ -123,6 +133,8 @@ export const DatasourceInputConfig: React.FunctionComponent<{
                               },
                             });
                           }}
+                          errors={inputVarsValidationResults.config![varName]}
+                          forceShowErrors={forceShowErrors}
                         />
                       </EuiFlexItem>
                     );
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 74b08f48df12d..a60ee4b9f6253 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,6 +19,7 @@ import {
   EuiSpacer,
 } from '@elastic/eui';
 import { DatasourceInput, DatasourceInputStream, RegistryInput } from '../../../../types';
+import { DatasourceInputValidationResults } from '../services';
 import { DatasourceInputConfig } from './datasource_input_config';
 import { DatasourceInputStreamConfig } from './datasource_input_stream_config';
 
@@ -32,7 +33,15 @@ 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);
 
@@ -122,6 +131,8 @@ export const DatasourceInputPanel: React.FunctionComponent<{
             packageInputVars={packageInput.vars}
             datasourceInput={datasourceInput}
             updateDatasourceInput={updateDatasourceInput}
+            inputVarsValidationResults={inputValidationResults}
+            forceShowErrors={forceShowErrors}
           />
           <EuiHorizontalRule margin="m" />
         </Fragment>
@@ -165,6 +176,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 3bf5b2bb4c0f0..64829c6d08cd4 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
@@ -16,14 +16,22 @@ import {
   EuiButtonEmpty,
 } from '@elastic/eui';
 import { DatasourceInputStream, RegistryStream, RegistryVarsEntry } from '../../../../types';
-import { isAdvancedVar } from '../services';
+import { isAdvancedVar, DatasourceConfigValidationResults } 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);
 
@@ -83,6 +91,8 @@ export const DatasourceInputStreamConfig: React.FunctionComponent<{
                       },
                     });
                   }}
+                  errors={inputStreamValidationResults.config![varName]}
+                  forceShowErrors={forceShowErrors}
                 />
               </EuiFlexItem>
             );
@@ -125,6 +135,8 @@ export const DatasourceInputStreamConfig: React.FunctionComponent<{
                               },
                             });
                           }}
+                          errors={inputStreamValidationResults.config![varName]}
+                          forceShowErrors={forceShowErrors}
                         />
                       </EuiFlexItem>
                     );
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 bcb99eed88ac0..846a807f9240d 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 e5f18e1449d1b..3bfca75668911 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 23d0f3317a667..7815ab9cd1d6e 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 44e5bfa41cb9b..d99f0712db3c3 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.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 0000000000000..3867ace9a0620
--- /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,211 @@
+/*
+ * 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>;
+}
+
+// Returns validation information for a given datasource configuration and package info
+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
+  ) {
+    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
+    inputValidationResults.config = Object.entries(input.config || {}).reduce(
+      (results, [name, configEntry]) => {
+        results[name] = input.enabled
+          ? validateDatasourceConfig(configEntry, inputVarsByName[name])
+          : null;
+        return results;
+      },
+      {} as ValidationEntry
+    );
+
+    // Validate each input stream with config fields
+    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] = stream.enabled
+            ? validateDatasourceConfig(configEntry, streamVarsByName[name])
+            : null;
+          return results;
+        },
+        {} as ValidationEntry
+      );
+
+      inputValidationResults.streams![stream.id] = streamValidationResults;
+    });
+
+    validationResults.inputs[input.type] = inputValidationResults;
+  });
+
+  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.type === 'yaml') {
+    try {
+      parsedValue = safeLoad(value);
+    } catch (e) {
+      errors.push(
+        i18n.translate('xpack.ingestManager.datasourceValidation.invalidYamlFormatErrorMessage', {
+          defaultMessage: 'Invalid YAML format',
+        })
+      );
+    }
+  }
+
+  if (varDef.required) {
+    if (varDef.multi) {
+      if (parsedValue && !Array.isArray(parsedValue)) {
+        errors.push(
+          i18n.translate('xpack.ingestManager.datasourceValidation.invalidArrayErrorMessage', {
+            defaultMessage: 'Invalid format',
+          })
+        );
+      }
+      if (!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,
+            },
+          })
+        );
+      }
+    } else {
+      if (parsedValue === undefined || (typeof parsedValue === 'string' && !parsedValue)) {
+        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
+    | ValidationEntry
+) => {
+  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 b45beef4a8b5e..95c78e2ee2609 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
@@ -13,13 +13,13 @@ import {
   EuiFlexGroup,
   EuiFlexItem,
   EuiFormRow,
-  EuiFieldText,
   EuiButtonEmpty,
   EuiSpacer,
   EuiEmptyPrompt,
   EuiText,
   EuiButton,
   EuiComboBox,
+  EuiCallOut,
 } from '@elastic/eui';
 import {
   AgentConfig,
@@ -28,21 +28,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,54 +97,54 @@ 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>
+          <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>
+          <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>
       <EuiSpacer size="m" />
@@ -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 0aa08602e4d4d..5ebd1300baf65 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 a59fb06145a3a..de1bf42ac6490 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,