feat(dataproc): add connection cluster to jobs

Author: james-zhou-inspire11

src/services/dataprocCluster/connections.ts

@@ -0,0 +1,49 @@
+import isEmpty from 'lodash/isEmpty'
+import { ServiceConnection } from '@cloudgraph/sdk'
+import { RawGcpDataprocCluster } from './data'
+import services from '../../enums/services'
+
+export default ({
+  service,
+  data,
+  region,
+}: {
+  service: RawGcpDataprocCluster
+  data: { name: string; data: { [property: string]: any[] } }[]
+  region: string
+}): {
+  [property: string]: ServiceConnection[]
+} => {
+  const { id } = service
+  const connections: ServiceConnection[] = []
+
+  /**
+   * Find Dataproc Jobs
+   */
+  const jobs: {
+    name: string
+    data: { [property: string]: any[] }
+  } = data.find(({ name }) => name === services.dataprocJob)
+
+  if (jobs?.data?.[region]) {
+
+    const filtered = jobs.data[region].filter(
+      ({ placement }) => placement.clusterUuid === id
+    )
+    if (!isEmpty(filtered)) {
+      for (const { id } of filtered) {
+        connections.push({
+          id,
+          resourceType: services.dataprocJob,
+          relation: 'child',
+          field: 'dataprocJobs',
+        })
+      }
+    }
+  }
+
+  const result = {
+    [id]: connections,
+  }
+  return result
+}

src/services/dataprocCluster/data.ts

@@ -1,4 +1,3 @@
-import cuid from 'cuid'
 import { ClusterControllerClient } from '@google-cloud/dataproc'
 import CloudGraph from '@cloudgraph/sdk'
 import groupBy from 'lodash/groupBy'
@@ -11,7 +10,7 @@ const lt = { ...gcpLoggerText }
 const { logger } = CloudGraph
 const serviceName = 'Dataproc Cluster'
 
-export interface RawGcpDataprocCluster extends Omit<google.cloud.dataproc.v1.ICluster, 'projectId' | 'labels'> {
+export interface RawGcpDataprocCluster extends Omit<google.cloud.dataproc.v1.ICluster, 'projectId' | 'clusterUuid' | 'labels'> {
   id: string
   region: string
   projectId: string
@@ -40,11 +39,11 @@ export default async ({
         })
 
         const iterable =  dataprocClient.listClustersAsync({ projectId, region })
-        for await (const { labels, ...response } of iterable) {
+        for await (const { clusterUuid, labels, ...response } of iterable) {
           if (response) {
             clusterList.push({
               ...response,
-              id: cuid(),
+              id: clusterUuid,
               projectId,
               region,
               Labels: labels,

src/services/dataprocCluster/format.ts

@@ -238,7 +238,6 @@ export default ({
     Labels,
     status = {},
     statusHistory = [],
-    clusterUuid,
     metrics = {},
   } = service
 
@@ -251,7 +250,6 @@ export default ({
     labels: formatLabelsFromMap(Labels),
     status: formatClusterStatus(status),
     statusHistory: statusHistory?.map(history => formatClusterStatus(history)),
-    clusterUuid,
     hdfsMetrics: formatKeyValueMap(metrics?.hdfsMetrics || {}),
     yarnMetrics: formatKeyValueMap(metrics?.yarnMetrics || {}),
   }

src/services/dataprocCluster/index.ts

@@ -2,12 +2,15 @@ import {Service} from '@cloudgraph/sdk'
 import BaseService from '../base'
 import format from './format'
 import getData from './data'
+import getConnections from './connections'
 import mutation from './mutation'
 
 export default class GcpDataprocCluster extends BaseService implements Service {
   format = format.bind(this)
 
   getData = getData.bind(this)
 
+  getConnections = getConnections.bind(this)
+
   mutation = mutation
 }

src/services/dataprocCluster/schema.graphql

@@ -194,8 +194,8 @@ type gcpDataprocCluster implements gcpBaseResource
     labels: [gcpRawLabel]
     status: gcpDataprocClusterStatus
     statusHistory: [gcpDataprocClusterStatus]
-    clusterUuid: String @search(by: [hash, regexp])
     hdfsMetrics: [gcpDataprocClusterMetric]
     yarnMetrics: [gcpDataprocClusterMetric]
+    dataprocJobs: [gcpDataprocJob] @hasInverse(field: dataprocClusters)
     project: [gcpProject] @hasInverse(field: dataprocClusters)
   }

src/services/dataprocJob/schema.graphql

@@ -173,5 +173,6 @@ type gcpDataprocJob implements gcpBaseResource
     schedulingMaxFailuresTotal: Int @search
     done: Boolean @search
     labels: [gcpRawLabel]
+    dataprocClusters: [gcpDataprocCluster] @hasInverse(field: dataprocJobs)
     project: [gcpProject] @hasInverse(field: dataprocJobs)
   }

src/types/generated.ts

@@ -876,8 +876,8 @@ export type GcpDataprocAutoscalingPolicy = GcpBaseResource & {
 };
 
 export type GcpDataprocCluster = GcpBaseResource & {
-  clusterUuid?: Maybe<Scalars['String']>;
   config?: Maybe<GcpDataprocClusterConfig>;
+  dataprocJobs?: Maybe<Array<Maybe<GcpDataprocJob>>>;
   hdfsMetrics?: Maybe<Array<Maybe<GcpDataprocClusterMetric>>>;
   labels?: Maybe<Array<Maybe<GcpRawLabel>>>;
   project?: Maybe<Array<Maybe<GcpProject>>>;
@@ -1029,6 +1029,7 @@ export type GcpDataprocInstanceClusterConfigAccelerator = {
 };
 
 export type GcpDataprocJob = GcpBaseResource & {
+  dataprocClusters?: Maybe<Array<Maybe<GcpDataprocCluster>>>;
   done?: Maybe<Scalars['Boolean']>;
   driverControlFilesUri?: Maybe<Scalars['String']>;
   driverOutputResourceUri?: Maybe<Scalars['String']>;