[Ingest Manager] Support both zip & tar archives from Registry

package.json

@@ -142,6 +142,7 @@
     "@kbn/test-subj-selector": "0.2.1",
     "@kbn/ui-framework": "1.0.0",
     "@kbn/ui-shared-deps": "1.0.0",
+    "@types/yauzl": "^2.9.1",
     "JSONStream": "1.3.5",
     "abortcontroller-polyfill": "^1.4.0",
     "accept": "3.0.2",

x-pack/plugins/ingest_manager/server/services/epm/registry/cache.ts

@@ -8,4 +8,12 @@ const cache: Map<string, Buffer> = new Map();
 export const cacheGet = (key: string) => cache.get(key);
 export const cacheSet = (key: string, value: Buffer) => cache.set(key, value);
 export const cacheHas = (key: string) => cache.has(key);
-export const getCacheKey = (key: string) => key + '.tar.gz';
+
+const archiveKeyCache: Map<string, string> = new Map();
+const stableKey = ({ name, version }: { name: string; version: string }) => `${name}-${version}`;
+
+export const getArchiveKey = (name: string, version: string) =>
+  archiveKeyCache.get(stableKey({ name, version }));
+
+export const setArchiveKey = (name: string, version: string, location: string) =>
+  archiveKeyCache.set(stableKey({ name, version }), location);

x-pack/plugins/ingest_manager/server/services/epm/registry/extract.ts

@@ -5,6 +5,7 @@
  */
 
 import tar from 'tar';
+import yauzl from 'yauzl';
 import { bufferToStream, streamToBuffer } from './streams';
 
 export interface ArchiveEntry {
@@ -30,3 +31,40 @@ export async function untarBuffer(
     deflatedStream.pipe(inflateStream);
   });
 }
+
+export async function unzipBuffer(
+  buffer: Buffer,
+  filter = (entry: ArchiveEntry): boolean => true,
+  onEntry = (entry: ArchiveEntry): void => {}
+): Promise<void> {
+  const zipfile = await yauzlFromBuffer(buffer, { lazyEntries: true });
+  zipfile.readEntry();
+  zipfile.on('entry', async (entry: yauzl.Entry) => {
+    const path = entry.fileName;
+    if (!filter({ path })) return zipfile.readEntry();
+
+    const entryBuffer = await getZipReadStream(zipfile, entry).then(streamToBuffer);
+    onEntry({ buffer: entryBuffer, path });
+    zipfile.readEntry();
+  });
+  return new Promise((resolve, reject) => zipfile.on('end', resolve).on('error', reject));
+}
+
+function yauzlFromBuffer(buffer: Buffer, opts: yauzl.Options): Promise<yauzl.ZipFile> {
+  return new Promise((resolve, reject) =>
+    yauzl.fromBuffer(buffer, opts, (err?: Error, handle?: yauzl.ZipFile) =>
+      err ? reject(err) : resolve(handle)
+    )
+  );
+}
+
+function getZipReadStream(
+  zipfile: yauzl.ZipFile,
+  entry: yauzl.Entry
+): Promise<NodeJS.ReadableStream> {
+  return new Promise((resolve, reject) =>
+    zipfile.openReadStream(entry, (err?: Error, readStream?: NodeJS.ReadableStream) =>
+      err ? reject(err) : resolve(readStream)
+    )
+  );
+}

x-pack/plugins/ingest_manager/server/services/epm/registry/index.ts

@@ -17,8 +17,8 @@ import {
   RegistrySearchResults,
   RegistrySearchResult,
 } from '../../../types';
-import { cacheGet, cacheSet, getCacheKey, cacheHas } from './cache';
-import { ArchiveEntry, untarBuffer } from './extract';
+import { cacheGet, cacheSet, cacheHas, getArchiveKey, setArchiveKey } from './cache';
+import { ArchiveEntry, untarBuffer, unzipBuffer } from './extract';
 import { fetchUrl, getResponse, getResponseStream } from './requests';
 import { streamToBuffer } from './streams';
 import { getRegistryUrl } from './registry_url';
@@ -182,17 +182,20 @@ async function extract(
   onEntry: (entry: ArchiveEntry) => void
 ) {
   const archiveBuffer = await getOrFetchArchiveBuffer(pkgName, pkgVersion);
+  const archiveKey = getArchiveKey(pkgName, pkgVersion);
+  // shouldn't be possible since getOrFetchArchiveBuffer -> fetchArchiveBuffer sets the key
+  if (!archiveKey) throw new Error('no archive key');
+  const isZip = archiveKey.endsWith('.zip');
+  const libExtract = isZip ? unzipBuffer : untarBuffer;
 
-  return untarBuffer(archiveBuffer, filter, onEntry);
+  return libExtract(archiveBuffer, filter, onEntry);
 }
 
 async function getOrFetchArchiveBuffer(pkgName: string, pkgVersion: string): Promise<Buffer> {
-  // assume .tar.gz for now. add support for .zip if/when we need it
-  const key = getCacheKey(`${pkgName}-${pkgVersion}`);
-  let buffer = cacheGet(key);
+  const key = getArchiveKey(pkgName, pkgVersion);
+  let buffer = key && cacheGet(key);
   if (!buffer) {
     buffer = await fetchArchiveBuffer(pkgName, pkgVersion);
-    cacheSet(key, buffer);
   }
 
   if (buffer) {
@@ -203,16 +206,21 @@ async function getOrFetchArchiveBuffer(pkgName: string, pkgVersion: string): Pro
 }
 
 export async function ensureCachedArchiveInfo(name: string, version: string) {
-  const pkgkey = getCacheKey(`${name}-${version}`);
-  if (!cacheHas(pkgkey)) {
+  const pkgkey = getArchiveKey(name, version);
+  if (!pkgkey || !cacheHas(pkgkey)) {
     await getArchiveInfo(name, version);
   }
 }
 
 async function fetchArchiveBuffer(pkgName: string, pkgVersion: string): Promise<Buffer> {
   const { download: archivePath } = await fetchInfo(pkgName, pkgVersion);
-  const registryUrl = getRegistryUrl();
-  return getResponseStream(`${registryUrl}${archivePath}`).then(streamToBuffer);
+  const archiveUrl = `${getRegistryUrl()}${archivePath}`;
+  const buffer = await getResponseStream(archiveUrl).then(streamToBuffer);
+
+  setArchiveKey(pkgName, pkgVersion, archivePath);
+  cacheSet(archivePath, buffer);
+
+  return buffer;
 }
 
 export function getAsset(key: string) {