FAI-8012 - Octopus should create Deploy->Artifact->Commit connections if possible (#1173)

willmarks · web-flow · commit 83fd02bff1af · 2023-10-17T12:06:39.000-07:00
diff --git a/destinations/airbyte-faros-destination/resources/source-specific-configs/octopus.json b/destinations/airbyte-faros-destination/resources/source-specific-configs/octopus.json
@@ -0,0 +1,24 @@
+{
+  "title": "Octopus",
+  "description": "Configuration options that apply to records generated by the Octopus Source.",
+  "type": "object",
+  "oneOf": [
+    {
+      "type": "object",
+      "title": "Configuration",
+      "properties": {
+        "source_type": {
+          "type": "string",
+          "const": "Octopus",
+          "order": 0
+        },
+        "vcs_source": {
+          "type": "string",
+          "title": "VCS Source",
+          "description": "The VCS source for the VCS information captured in Octopus",
+          "default": "GitHub"
+        }
+      }
+    }
+  ]
+}
diff --git a/destinations/airbyte-faros-destination/resources/source-specific-configs/spec.json b/destinations/airbyte-faros-destination/resources/source-specific-configs/spec.json
@@ -51,6 +51,9 @@
         "notion": {
           "$ref": "notion.json"
         },
+        "octopus": {
+          "$ref": "octopus.json"
+        },
         "opsgenie": {
           "$ref": "opsgenie.json"
         },
diff --git a/destinations/airbyte-faros-destination/src/converters/octopus/common.ts b/destinations/airbyte-faros-destination/src/converters/octopus/common.ts
@@ -1,6 +1,10 @@
 import {AirbyteRecord} from 'faros-airbyte-cdk';
 
-import {Converter} from '../converter';
+import {Converter, StreamContext} from '../converter';
+
+interface OctopusConfig {
+  vcs_source?: string;
+}
 
 /** Octopus converter base */
 export abstract class OctopusConverter extends Converter {
@@ -9,4 +13,12 @@ export abstract class OctopusConverter extends Converter {
   id(record: AirbyteRecord): any {
     return record?.record?.data?.Id;
   }
+
+  protected octopusConfig(ctx: StreamContext): OctopusConfig {
+    return ctx.config.source_specific_configs?.octopus ?? {};
+  }
+
+  protected vcsSource(ctx: StreamContext): string | undefined {
+    return this.octopusConfig(ctx)?.vcs_source;
+  }
 }
diff --git a/destinations/airbyte-faros-destination/src/converters/octopus/deployments.ts b/destinations/airbyte-faros-destination/src/converters/octopus/deployments.ts
@@ -1,10 +1,23 @@
-import {AirbyteRecord} from 'faros-airbyte-cdk';
+import {AirbyteLogger, AirbyteRecord} from 'faros-airbyte-cdk';
 import {Utils} from 'faros-js-client';
+import GitUrlParse from 'git-url-parse';
 
 import {Common} from '../common/common';
 import {DestinationModel, DestinationRecord, StreamContext} from '../converter';
 import {OctopusConverter} from './common';
 
+interface ArtifactVCSInfo {
+  artifactUid: string;
+  sha?: string;
+  repository: {
+    name: string;
+    organization: {
+      uid: string;
+      source: string;
+    };
+  };
+}
+
 export class Deployments extends OctopusConverter {
   readonly destinationModels: ReadonlyArray<DestinationModel> = [
     'cicd_Deployment',
@@ -18,10 +31,15 @@ export class Deployments extends OctopusConverter {
     const source = this.streamName.source;
     const deployment = record.record.data;
 
+    const deploymentKey = {
+      uid: deployment.Id,
+      source,
+    };
+
     res.push({
       model: 'cicd_Deployment',
       record: {
-        uid: deployment.Id,
+        ...deploymentKey,
         application: Common.computeApplication(deployment.ProjectName),
         url: deployment.Links?.Self,
         requestedAt: Utils.toDate(deployment.Task?.QueueTime),
@@ -32,13 +50,206 @@ export class Deployments extends OctopusConverter {
           deployment.Task?.State,
           deployment.Task?.ErrorMessage
         ),
-        source,
       },
     });
 
+    const vcsInfo = this.getVCSInfo(deployment, ctx);
+
+    for (const vcs of vcsInfo) {
+      const artifactKey = {
+        uid: vcs.artifactUid,
+        repository: {
+          uid: vcs.repository.name,
+          organization: vcs.repository.organization,
+        },
+      };
+
+      res.push({
+        model: 'cicd_ArtifactDeployment',
+        record: {
+          artifact: artifactKey,
+          deployment: deploymentKey,
+        },
+      });
+
+      if (vcs.sha) {
+        // Only instantiate artifact if we can link all the way to commit
+        res.push({
+          model: 'cicd_Artifact',
+          record: artifactKey,
+        });
+        res.push({
+          model: 'cicd_ArtifactCommitAssociation',
+          record: {
+            artifact: artifactKey,
+            commit: {
+              sha: vcs.sha,
+              repository: vcs.repository,
+            },
+          },
+        });
+      }
+    }
+
     return res;
   }
 
+  private getVCSInfo(deployment: any, ctx: StreamContext): ArtifactVCSInfo[] {
+    const deployId = deployment.Id;
+    const logger = ctx.logger;
+
+    let commitSha: string;
+    let repoName: string;
+    let orgUid: string;
+    let orgSource: string = this.vcsSource(ctx);
+
+    // Attempt to first retrieve VCS information from explicit deployment variables
+    for (const v of deployment.Variables ?? []) {
+      if (v.Name === 'VCS_COMMIT') {
+        commitSha = this.getVarValue(deployId, v, logger);
+      } else if (v.name === 'VCS_REPO') {
+        repoName = this.getVarValue(deployId, v, logger);
+      } else if (v.name === 'VCS_ORG') {
+        orgUid = this.getVarValue(deployId, v, logger);
+      } else if (v.name === 'VCS_SOURCE') {
+        orgSource = this.getVarValue(deployId, v, logger);
+      }
+    }
+
+    // If all information provided as variables return it
+    if (commitSha && repoName && orgUid) {
+      return [
+        {
+          artifactUid: commitSha,
+          sha: commitSha,
+          repository: {
+            name: repoName,
+            organization: {
+              uid: orgUid,
+              source: orgSource,
+            },
+          },
+        },
+      ];
+    }
+
+    const vcsInfo: ArtifactVCSInfo[] = [];
+    const hasChangeArray =
+      deployment.Changes &&
+      Array.isArray(deployment.Changes) &&
+      deployment.Changes.length > 0;
+
+    if (hasChangeArray) {
+      // First try to construct the fallback repo from explicit deployment variables
+      let fallbackRepoInfo: ArtifactVCSInfo['repository'];
+      if (repoName && orgUid) {
+        fallbackRepoInfo = {
+          name: repoName,
+          organization: {
+            uid: orgUid,
+            source: orgSource,
+          },
+        };
+      }
+
+      // If fallback not set attempt to retrieve it from repo property in process
+      if (!fallbackRepoInfo) {
+        for (const step of deployment.Process?.Steps ?? []) {
+          for (const action of step.Actions ?? []) {
+            const repo = action.Properties?.['repo'];
+            if (repo) {
+              fallbackRepoInfo = this.getRepoInfoFromURL(deployId, repo, ctx);
+              break;
+            }
+          }
+          if (fallbackRepoInfo) {
+            break;
+          }
+        }
+      }
+
+      // Assume last change was the one deployed
+      const change = deployment.Changes.at(-1);
+
+      // Retrieve repo information from Commits
+      for (const commit of change.Commits ?? []) {
+        const artifactUid = change.Version;
+        const sha = commit.Id;
+        const repository =
+          this.getRepoInfoFromURL(deployId, commit.LinkUrl, ctx) ??
+          fallbackRepoInfo;
+
+        if (repository) {
+          vcsInfo.push({artifactUid, sha, repository});
+        }
+      }
+
+      // Commits did not yield any VCS info, use BuildInformation
+      if (!vcsInfo.length) {
+        for (const buildInfo of change.BuildInformation ?? []) {
+          const artifactUid = `${buildInfo.PackageId}:${buildInfo.Version}`;
+          const sha = buildInfo.VcsCommitNumber;
+          const repository =
+            this.getRepoInfoFromURL(deployId, buildInfo.VcsRoot, ctx) ??
+            fallbackRepoInfo;
+
+          if (repository) {
+            vcsInfo.push({artifactUid, sha, repository});
+          }
+        }
+      }
+
+      // If still no info create a dummy artifact using Change.Version
+      if (!vcsInfo.length && fallbackRepoInfo) {
+        vcsInfo.push({
+          artifactUid: change.Version,
+          repository: fallbackRepoInfo,
+        });
+      }
+    }
+
+    if (!vcsInfo.length) {
+      logger.warn(`Could not retrieve VCS info for deployment: ${deployId}`);
+    }
+
+    return vcsInfo;
+  }
+
+  private getRepoInfoFromURL(
+    deploymentId: string,
+    url: string,
+    ctx: StreamContext
+  ): ArtifactVCSInfo['repository'] {
+    try {
+      const parsedUrl = GitUrlParse(url);
+      return {
+        name: parsedUrl.name.toLowerCase(),
+        organization: {
+          uid: parsedUrl.owner.toLowerCase(),
+          source: this.vcsSource(ctx),
+        },
+      };
+    } catch (err: any) {
+      ctx.logger.warn(
+        `Unable to parse VCS information for deployment ${deploymentId} from repo url: ${url}`
+      );
+    }
+  }
+
+  private getVarValue(
+    deploymentId: string,
+    variable: {Name: string; Value: string},
+    logger: AirbyteLogger
+  ): string | undefined {
+    if (variable.Value) {
+      return variable.Value;
+    }
+    logger.warn(
+      `Deployment [${deploymentId}] had a null value for ${variable.Name} variable`
+    );
+    return undefined;
+  }
+
   /**
    * Octopus task statuses include:
    * Canceled, Cancelling, Executing, Failed, Queued, Success, TimedOut
