diff --git a/x-pack/legacy/plugins/epm/server/lib/elasticsearch/ingest_pipeline/ingest_pipelines.ts b/x-pack/legacy/plugins/epm/server/lib/elasticsearch/ingest_pipeline/ingest_pipelines.ts
index 216abbbd28cd4..f59d9fcf35d5b 100644
--- a/x-pack/legacy/plugins/epm/server/lib/elasticsearch/ingest_pipeline/ingest_pipelines.ts
+++ b/x-pack/legacy/plugins/epm/server/lib/elasticsearch/ingest_pipeline/ingest_pipelines.ts
@@ -55,10 +55,32 @@ async function installPipeline({
 }): Promise<AssetReference> {
   const buffer = Registry.getAsset(path);
   const parts = Registry.pathParts(path);
+  const extension = getExtension(path);
   const id = path.replace(/\W/g, '_'); // TODO: replace with "real" pipeline id
   const pipeline = buffer.toString('utf8');
 
-  await callCluster('ingest.putPipeline', { id, body: pipeline });
+  const callClusterParams: {
+    method: string;
+    path: string;
+    ignore: number[];
+    body: any;
+    headers?: any;
+  } = {
+    method: 'PUT',
+    path: `/_ingest/pipeline/${id}`,
+    ignore: [404],
+    body: pipeline,
+  };
+  if (extension === 'yml') {
+    callClusterParams.headers = { ['Content-Type']: 'application/yaml' };
+  }
+
+  // This uses the catch-all endpoint 'transport.request' because we have to explicitly
+  // set the Content-Type header above for sending yml data. Setting the headers is not
+  // exposed in the convenience endpoint 'ingest.putPipeline' of elasticsearch-js-legacy
+  // which we could otherwise use.
+  // See src/core/server/elasticsearch/api_types.ts for available endpoints.
+  await callCluster('transport.request', callClusterParams);
 
   return { id, type: parts.type };
 }
@@ -66,3 +88,8 @@ async function installPipeline({
 const isDirectory = ({ path }: Registry.ArchiveEntry) => path.endsWith('/');
 const isPipeline = ({ path }: Registry.ArchiveEntry) =>
   !isDirectory({ path }) && Registry.pathParts(path).type === ElasticsearchAssetType.ingestPipeline;
+
+const getExtension = (path: string): string => {
+  const splitPath = path.split('.');
+  return splitPath[splitPath.length - 1];
+};
diff --git a/x-pack/test/epm_api_integration/apis/data_sources.ts b/x-pack/test/epm_api_integration/apis/data_sources.ts
new file mode 100644
index 0000000000000..22ce8c1026317
--- /dev/null
+++ b/x-pack/test/epm_api_integration/apis/data_sources.ts
@@ -0,0 +1,105 @@
+/*
+ * 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.
+ */
+
+/* eslint-disable no-console */
+/* tslint:disable */
+
+import { readFileSync } from 'fs';
+import path from 'path';
+import expect from '@kbn/expect';
+import ServerMock from 'mock-http-server';
+import { FtrProviderContext } from '../../api_integration/ftr_provider_context';
+
+export default function({ getService }: FtrProviderContext) {
+  const supertest = getService('supertest');
+  describe('data source installation', () => {
+    const registryMock = new ServerMock({ host: 'localhost', port: 6666 });
+    beforeEach(async () => {
+      registryMock.start(() => {});
+      const packageResponse = readFileSync(
+        path.join(__dirname, '/fixtures/packages/package/yamlpipeline_1.0.0')
+      ).toString();
+      const fileResponse = readFileSync(
+        path.join(__dirname, '/fixtures/packages/epr/yamlpipeline_1.0.0.tar.gz')
+      );
+      registryMock.on({
+        method: 'GET',
+        path: '/package/yamlpipeline-1.0.0',
+        reply: {
+          status: 200,
+          headers: { 'content-type': 'application/json' },
+          body: packageResponse,
+        },
+      });
+
+      registryMock.on({
+        method: 'GET',
+        path: '/epr/yamlpipeline/yamlpipeline-1.0.0.tar.gz',
+        reply: {
+          status: 200,
+          headers: { 'content-type': 'application/gzip' },
+          body: fileResponse,
+        },
+      });
+
+      const installPackage = async () => {
+        const response = await supertest
+          .get('/api/epm/install/yamlpipeline-1.0.0')
+          .set('kbn-xsrf', 'xxx')
+          .expect(200);
+        return response.body;
+      };
+      // commment for debugging
+      await installPackage();
+
+      // uncomment for debugging
+      // const packageInstallResponse = await installPackage();
+      // console.log('packageInstallResponse is: ', packageInstallResponse);
+      // console.log(
+      //   'requests are',
+      //   registryMock.requests().map(r => r.url)
+      // );
+    });
+    afterEach(() => {
+      registryMock.stop(() => {});
+    });
+    it('test setup works', async () => {
+      const readPackageSavedObject = async () => {
+        const response = await supertest
+          .get('/api/saved_objects/epm-package/yamlpipeline-1.0.0')
+          .expect(200);
+        return response.body;
+      };
+      const savedObjectResponse = await readPackageSavedObject();
+      expect(savedObjectResponse.id).to.be('yamlpipeline-1.0.0');
+    });
+    it('works with a package containing only yml format ingest pipelines', async () => {
+      const createDataSource = async () => {
+        const response = await supertest
+          .get('/api/epm/datasource/install/yamlpipeline-1.0.0')
+          .expect(200);
+        return response.body;
+      };
+
+      const readDataSourceSavedObject = async () => {
+        const response = await supertest
+          .get('/api/saved_objects/epm-datasource/yamlpipeline-1.0.0')
+          .expect(200);
+        return response.body;
+      };
+
+      // comment for debugging
+      await createDataSource();
+
+      // uncomment for debugging
+      // const createDataSourceResponse = await createDataSource();
+      // console.log('createDataSourceResponse is ', createDataSourceResponse);
+
+      const readDataSourceSavedObjectResponse = await readDataSourceSavedObject();
+      expect(readDataSourceSavedObjectResponse.id).to.be('yamlpipeline-1.0.0');
+    });
+  });
+}
diff --git a/x-pack/test/epm_api_integration/apis/fixtures/packages/epr/yamlpipeline_1.0.0.tar.gz b/x-pack/test/epm_api_integration/apis/fixtures/packages/epr/yamlpipeline_1.0.0.tar.gz
new file mode 100644
index 0000000000000..ca8695f111d02
Binary files /dev/null and b/x-pack/test/epm_api_integration/apis/fixtures/packages/epr/yamlpipeline_1.0.0.tar.gz differ
diff --git a/x-pack/test/epm_api_integration/apis/fixtures/packages/package/yamlpipeline_1.0.0 b/x-pack/test/epm_api_integration/apis/fixtures/packages/package/yamlpipeline_1.0.0
new file mode 100644
index 0000000000000..bf167c583aab3
--- /dev/null
+++ b/x-pack/test/epm_api_integration/apis/fixtures/packages/package/yamlpipeline_1.0.0
@@ -0,0 +1,32 @@
+{
+  "name": "yamlpipeline",
+  "title": "Yaml Pipeline package",
+  "version": "1.0.0",
+  "description": "This package contains a yaml pipeline.\n",
+  "type": "integration",
+  "categories": [
+    "logs"
+  ],
+  "requirement": {
+    "kibana": {}
+  },
+  "assets": [
+    "/package/yamlpipeline-1.0.0/manifest.yml",
+    "/package/yamlpipeline-1.0.0/dataset/log/manifest.yml",
+    "/package/yamlpipeline-1.0.0/dataset/log/elasticsearch/ingest-pipeline/pipeline-entry.yml",
+    "/package/yamlpipeline-1.0.0/dataset/log/elasticsearch/ingest-pipeline/pipeline-json.yml",
+    "/package/yamlpipeline-1.0.0/dataset/log/elasticsearch/ingest-pipeline/pipeline-plaintext.yml"
+  ],
+  "format_version": "1.0.0",
+  "datasets": [
+    {
+      "title": "Log Yaml pipeline",
+      "name": "log",
+      "release": "",
+      "type": "logs",
+      "ingest_pipeline": ""
+    }
+  ],
+  "download": "/epr/yamlpipeline/yamlpipeline-1.0.0.tar.gz",
+  "path": "/package/yamlpipeline-1.0.0"
+}
\ No newline at end of file
diff --git a/x-pack/test/epm_api_integration/apis/index.js b/x-pack/test/epm_api_integration/apis/index.js
index 3dc4624d15cf4..015d2d3fa30f3 100644
--- a/x-pack/test/epm_api_integration/apis/index.js
+++ b/x-pack/test/epm_api_integration/apis/index.js
@@ -11,5 +11,6 @@ export default function ({ loadTestFile }) {
     loadTestFile(require.resolve('./file'));
     loadTestFile(require.resolve('./template'));
     loadTestFile(require.resolve('./ilm'));
+    loadTestFile(require.resolve('./data_sources'));
   });
 }