Merge pull request #21 from platformercloud/hotfix/continue-import-with-fetch-failures

Continue import with fetch failures

Author: ChathurangaKCD

src/commands/import.ts

@@ -1,15 +1,9 @@
 import { flags } from '@oclif/command';
 import { cli } from 'cli-ux';
 import { Observable, of } from 'rxjs';
-import {
-  concatMap,
-  count,
-  filter,
-  mergeMap,
-  tap,
-  toArray,
-} from 'rxjs/operators';
+import { count, filter, mergeMap, tap, toArray } from 'rxjs/operators';
 import Command from '../base-command';
+import { ensureTargetNamespace } from '../modules/apps/environment';
 import { getClusterNamespaces, listClusters } from '../modules/cluster/api';
 import {
   getDefaultEnvironment,
@@ -29,7 +23,6 @@ import {
 import { writeHAR } from '../modules/util/fetch';
 import { tryValidateCommonFlags } from '../modules/util/validations';
 import chalk = require('chalk');
-import { ensureTargetNamespace } from '../modules/apps/environment';
 
 export default class Apply extends Command {
   static description =
@@ -129,20 +122,18 @@ export default class Apply extends Command {
     );
     try {
       try {
-        await of(...importGroups)
-          .pipe(
-            concatMap((group) => {
-              const manifestOfGroup = group.getManifests();
-              return applyManifests(
-                manifestOfGroup,
-                { orgId, projectId, envId },
-                {
-                  start: `Applying ${group.resourceTypes.description}`,
-                }
-              );
-            })
-          )
-          .toPromise();
+        for (const group of importGroups) {
+          const manifestOfGroup = group.getManifests();
+          await applyManifests(
+            manifestOfGroup,
+            { orgId, projectId, envId },
+            {
+              start: `Applying ${group.resourceTypes.description}`,
+            }
+          );
+          // if no apply failures occured, show warnings for fetch failures
+          displayFetchFailures(group);
+        }
       } catch (error) {
         // if error occurs, append msg to the running spinner
         cli.action.stop('Error occured');
@@ -315,10 +306,9 @@ async function printLogs(groups: ManifestImportGroup[]) {
     .pipe(
       filter((m) => m.state === ManifestState.SKIPPED),
       tap((manifest) => {
-        const subTree = cli.tree();
         const kind = manifest.manifest.kind;
         const name = manifest.manifest.metadata.name;
-        skippedTree.insert(`${kind} ${name}`, subTree);
+        skippedTree.insert(`${kind} ${name}`);
       }),
       count(),
       tap((count) => {
@@ -331,6 +321,19 @@ async function printLogs(groups: ManifestImportGroup[]) {
     .toPromise();
 }
 
+async function displayFetchFailures(group: ManifestImportGroup) {
+  if (group.fetchFailures.length === 0) return;
+  const tree = cli.tree();
+  group.fetchFailures.forEach((failure) => {
+    const subTree = cli.tree();
+    tree.insert(chalk.yellow(`${failure.message}`), subTree);
+    if (failure.cause) {
+      subTree.insert(failure.cause);
+    }
+  });
+  tree.display();
+}
+
 function printSummary(statusArr: ManifestState[]) {
   const manifestCount = statusArr.length;
   const summary = new Map<ManifestState, number>();

src/modules/cluster/api.ts

@@ -96,11 +96,10 @@ interface ClusterResourceQuery {
   apiVersion: string;
   kind: string;
 }
-interface ClusterResources {
+export interface ClusterResources {
   clusterID: string;
   payload: K8sObject[];
 }
-
 /**
  * Query any resource on the Cluster
  * Tip: use kubectl api-versions and kubectl api-resources to figure out the exact

src/modules/errors/api-error.ts

@@ -13,4 +13,11 @@ export default class APIError extends Error {
   get statusCode() {
     return this.response.status;
   }
+  async data() {
+    try {
+      return await this.response.json();
+    } catch (error) {
+      return null;
+    }
+  }
 }

src/modules/gitops/manifest-group.ts

@@ -1,5 +1,6 @@
-import { defer, Observable, of } from 'rxjs';
+import { defer, EMPTY, Observable, of } from 'rxjs';
 import {
+  catchError,
   filter,
   map,
   mergeAll,
@@ -9,7 +10,8 @@ import {
   takeUntil,
   tap,
 } from 'rxjs/operators';
-import { queryResource } from '../cluster/api';
+import { ClusterResources, queryResource } from '../cluster/api';
+import APIError from '../errors/api-error';
 import { ImportType, ResourceType } from './manifest-import-types';
 import {
   ManifestObject,
@@ -28,6 +30,7 @@ interface ResourceQuery {
 export class ManifestImportGroup {
   resourceTypes: ImportType;
   #manifests: ManifestObject[] = [];
+  fetchFailures: YamlFectchFailure[] = [];
   #observable: Observable<ManifestObject>;
   constructor(
     resourceTypes: ImportType,
@@ -38,6 +41,13 @@ export class ManifestImportGroup {
     this.resourceTypes = resourceTypes;
     this.#observable = of(...this.resourceTypes.types).pipe(
       mergeMap((r) => fetchResourcesOfType(query, r), 4),
+      filter((v): v is ClusterResources => {
+        if (v instanceof YamlFectchFailure) {
+          this.fetchFailures.push(v);
+          return false;
+        }
+        return true;
+      }),
       takeUntil(skippedStateNotifier),
       map((v) => v.payload),
       mergeAll(),
@@ -70,16 +80,65 @@ function shouldImport(v: K8sObject, sourceNS: string) {
 
 function fetchResourcesOfType(query: ResourceQuery, r: ResourceType) {
   const { orgId, projectId, clusterId, namespace } = query;
-  return defer(() =>
-    queryResource(
-      {
-        orgId,
-        projectId,
-        clusterId,
-        apiVersion: r.apiVersion,
-        kind: r.kind,
-      },
-      { namespace }
-    )
-  ).pipe(retry(2));
+  return defer(async () => {
+    try {
+      const res = await queryResource(
+        {
+          orgId,
+          projectId,
+          clusterId,
+          apiVersion: r.apiVersion,
+          kind: r.kind,
+        },
+        { namespace }
+      );
+      return res;
+    } catch (err) {
+      const data = err instanceof APIError ? await err.data() : null;
+      const cause = isClusterResourcesFailure(data) ? data.error : null;
+      throw new YamlFectchFailure(r, query, cause);
+    }
+  }).pipe(
+    retry(2),
+    catchError(async (err) => {
+      if (err instanceof YamlFectchFailure) return err;
+      return EMPTY;
+    })
+  );
+}
+
+export class YamlFectchFailure extends Error {
+  readonly resourceType: ResourceType;
+  readonly resourceQuery: ResourceQuery;
+  readonly cause: string | null;
+  constructor(
+    resourceType: ResourceType,
+    resourceQuery: ResourceQuery,
+    cause: string | null
+  ) {
+    super(
+      `Failed to fetch resources of kind : ${resourceType.kind} (${resourceType.apiVersion})`
+    );
+    this.cause = cause;
+    this.resourceType = resourceType;
+    this.resourceQuery = resourceQuery;
+  }
+}
+
+export interface ClusterResourcesFailure {
+  clusterID: string;
+  error: string;
+  identifier: 'resource:get';
+  requestID: string;
+  status: 'fail';
+  type: 'query';
+}
+export function isClusterResourcesFailure(
+  data: any
+): data is ClusterResourcesFailure {
+  if (typeof data !== 'object' || data === null) return false;
+  if (!['status', 'error'].every((prop) => typeof data[prop] === 'string')) {
+    return false;
+  }
+  return data['status'] === 'fail';
 }

src/modules/util/fetch.ts

@@ -4,9 +4,11 @@ import nodeFetch from 'node-fetch';
 import { homedir } from 'os';
 // @ts-ignore
 import { createHarLog, withHar } from 'node-fetch-har';
+import { cli } from 'cli-ux';
 
 const homeDir = homedir();
-const LOG_REQUESTS = false;
+const LOG_REQUESTS =
+  process.env.DEBUG === '*' || process.env.DEBUG === 'REQUESTS';
 
 export const { fetch, writeHAR } = (function () {
   if (!LOG_REQUESTS)
@@ -20,10 +22,21 @@ export const { fetch, writeHAR } = (function () {
   const fetch = withHar(nodeFetch, {
     onHarEntry: (entry: any) => har.push(entry),
   });
-
   function writeHAR(fileName = `har-${new Date().toISOString()}.har`) {
-    const filePath = path.join(homeDir, 'logs', fileName);
-    fs.writeFileSync(filePath, JSON.stringify(createHarLog(har), null, 2));
+    writeToHAR(fileName, har);
   }
   return { fetch, writeHAR };
 })();
+
+function writeToHAR(fileName: string, logs: any[]) {
+  const folderPath = path.join(homeDir, 'platformer-logs');
+  const filePath = path.join(folderPath, fileName);
+  try {
+    fs.mkdirSync(folderPath, { recursive: true });
+  } catch (error) {}
+  try {
+    fs.writeFileSync(filePath, JSON.stringify(createHarLog(logs), null, 2));
+  } catch (error) {
+    cli.error(`Failed to write to ${filePath}`);
+  }
+}