feat(dataproc): add workflow template

Author: james-zhou-inspire11

src/enums/schemasMap.ts

@@ -43,5 +43,6 @@ export default {
   [services.dataprocCluster]: 'gcpDataprocCluster',
   [services.dataprocAutoscalingPolicy]: 'gcpDataprocAutoscalingPolicy',
   [services.dataprocJob]: 'gcpDataprocJob',
+  [services.dataprocWorkflowTemplate]: 'gcpDataprocWorkflowTemplate',
   tag: 'gcpTag',
 }

src/enums/serviceMap.ts

@@ -39,7 +39,7 @@ import GcpComputeProject from '../services/computeProject'
 import GcpDataprocCluster from '../services/dataprocCluster'
 import GcpDataprocAutoscalingPolicy from '../services/dataprocAutoscalingPolicy'
 import GcpDataprocJob from '../services/dataprocJob'
-
+import GcpDataprocWorkflowTemplate from '../services/dataprocWorkflowTemplate'
 /**
  * serviceMap is an object that contains all currently supported services
  * serviceMap is used by the serviceFactory to produce instances of service classes
@@ -84,5 +84,6 @@ export default {
   [services.dataprocCluster]: GcpDataprocCluster,
   [services.dataprocAutoscalingPolicy]: GcpDataprocAutoscalingPolicy,
   [services.dataprocJob]: GcpDataprocJob,
+  [services.dataprocWorkflowTemplate]: GcpDataprocWorkflowTemplate,
   tag: GcpTag,
 }

src/enums/services.ts

@@ -31,6 +31,7 @@ export default {
   dataprocCluster: 'dataprocClusters',
   dataprocAutoscalingPolicy: 'dataprocAutoscalingPolicies',
   dataprocJob: 'dataprocJobs',
+  dataprocWorkflowTemplate: 'dataprocWorkflowTemplates',
   // metastore: 'metastore',
   // pubsub: 'pubsub',
   // bigtable: 'bigtable',

src/services/dataprocCluster/connections.ts

@@ -2,6 +2,7 @@ import isEmpty from 'lodash/isEmpty'
 import { ServiceConnection } from '@cloudgraph/sdk'
 import { RawGcpDataprocCluster } from './data'
 import services from '../../enums/services'
+import { RawGcpDataprocWorkflowTemplate } from '../dataprocWorkflowTemplate/data'
 
 export default ({
   service,
@@ -14,7 +15,7 @@ export default ({
 }): {
   [property: string]: ServiceConnection[]
 } => {
-  const { id } = service
+  const { id, Labels } = service
   const connections: ServiceConnection[] = []
 
   /**
@@ -26,7 +27,6 @@ export default ({
   } = data.find(({ name }) => name === services.dataprocJob)
 
   if (jobs?.data?.[region]) {
-
     const filtered = jobs.data[region].filter(
       ({ placement }) => placement.clusterUuid === id
     )
@@ -42,6 +42,35 @@ export default ({
     }
   }
 
+  /**
+   * Find Dataproc Workflow Template
+   */
+  const flatClusterLabels = Object.keys(Labels || {}).map(key => `${key}:${Labels[key]}`)
+
+  const templates: {
+    name: string
+    data: { [property: string]: any[] }
+  } = data.find(({ name }) => name === services.dataprocWorkflowTemplate)
+
+  if (templates?.data?.[region]) {
+    const filtered = templates.data[region]
+      .filter((template: RawGcpDataprocWorkflowTemplate) => !isEmpty(
+        Object.keys(template?.placement?.clusterSelector?.clusterLabels || {})
+          .map(key => `${key}:${template.placement.clusterSelector.clusterLabels[key]}`)
+          .filter(cluster => flatClusterLabels.includes(cluster))
+      )
+    )
+
+    for (const { id } of filtered) {
+      connections.push({
+        id,
+        resourceType: services.dataprocWorkflowTemplate,
+        relation: 'child',
+        field: 'dataprocWorkflowTemplates',
+      })
+    }
+  }
+
   const result = {
     [id]: connections,
   }

src/services/dataprocCluster/format.ts

@@ -172,7 +172,7 @@ const formatClusterConfigGke = ({
   }
 }
 
-const formatClusterConfig = ({
+export const formatClusterConfig = ({
   configBucket,
   tempBucket,
   gceClusterConfig = {},

src/services/dataprocCluster/schema.graphql

@@ -197,5 +197,6 @@ type gcpDataprocCluster implements gcpBaseResource
     hdfsMetrics: [gcpDataprocClusterMetric]
     yarnMetrics: [gcpDataprocClusterMetric]
     dataprocJobs: [gcpDataprocJob] @hasInverse(field: dataprocClusters)
+    dataprocWorkflowTemplates: [gcpDataprocWorkflowTemplate] @hasInverse(field: dataprocClusters)
     project: [gcpProject] @hasInverse(field: dataprocClusters)
   }

src/services/dataprocJob/format.ts

@@ -8,9 +8,9 @@ import {
   GcpDataprocJobYarnApplication,
   GcpDataprocPigJob,
   GcpDataprocPrestoJob,
-  GcpDataprocPysparkJob,
+  GcpDataprocPySparkJob,
   GcpDataprocSparkJob,
-  GcpDataprocspArkRJob,
+  GcpDataprocSparkRJob,
   GcpDataprocSparkSqlJob,
   GcpKeyValue,
 } from '../../types/generated'
@@ -109,7 +109,7 @@ const formatPysparkJob = ({
   archiveUris = [],
   properties = {},
   loggingConfig = {},
-}: google.cloud.dataproc.v1.IPySparkJob): GcpDataprocPysparkJob => {
+}: google.cloud.dataproc.v1.IPySparkJob): GcpDataprocPySparkJob => {
   return {
     mainPythonFileUri,
     args,
@@ -167,7 +167,7 @@ const formatSparkRJob = ({
   archiveUris = [],
   properties = {},
   loggingConfig = {},
-}: google.cloud.dataproc.v1.ISparkRJob): GcpDataprocspArkRJob => {
+}: google.cloud.dataproc.v1.ISparkRJob): GcpDataprocSparkRJob => {
   return {
     mainRFileUri,
     args,
@@ -256,7 +256,7 @@ export default ({
     placementClusterLabels: formatKeyValueMap(placement?.clusterLabels || {}),
     hadoopJob: formatHadoopJob(hadoopJob),
     sparkJob: formatSparkJob(sparkJob),
-    pysparkJob: formatPysparkJob(pysparkJob),
+    pySparkJob: formatPysparkJob(pysparkJob),
     hiveJob: formatHiveJob( hiveJob),
     pigJob: formatPigJob(pigJob),
     sparkRJob: formatSparkRJob(sparkRJob),

src/services/dataprocJob/index.ts

@@ -4,7 +4,7 @@ import format from './format'
 import getData from './data'
 import mutation from './mutation'
 
-export default class GcpDataprocCluster extends BaseService implements Service {
+export default class GcpDataprocJob extends BaseService implements Service {
   format = format.bind(this)
 
   getData = getData.bind(this)

src/services/dataprocJob/schema.graphql

@@ -58,7 +58,7 @@ type gcpDataprocSparkJob
     loggingConfig: [gcpKeyValue]
   }
 
-type gcpDataprocPysparkJob
+type gcpDataprocPySparkJob
   @generate(
     query: { get: true, query: true, aggregate: true }
     mutation: { add: true, delete: false }
@@ -103,7 +103,7 @@ type gcpDataprocPigJob
     loggingConfig: [gcpKeyValue]
   }
 
-type gcpDataprocspArkRJob
+type gcpDataprocSparkRJob
   @generate(
     query: { get: true, query: true, aggregate: true }
     mutation: { add: true, delete: false }
@@ -158,10 +158,10 @@ type gcpDataprocJob implements gcpBaseResource
     placementClusterLabels: [gcpKeyValue]
     hadoopJob: gcpDataprocHadoopJob
     sparkJob: gcpDataprocSparkJob
-    pysparkJob: gcpDataprocPysparkJob
+    pySparkJob: gcpDataprocPySparkJob
     hiveJob: gcpDataprocHiveJob
     pigJob: gcpDataprocPigJob
-    sparkRJob: gcpDataprocspArkRJob
+    sparkRJob: gcpDataprocSparkRJob
     sparkSqlJob: gcpDataprocSparkSqlJob
     prestoJob: gcpDataprocPrestoJob
     status: gcpDataprocJobStatus

src/services/dataprocWorkflowTemplate/data.ts

@@ -0,0 +1,61 @@
+import { WorkflowTemplateServiceClient } from '@google-cloud/dataproc'
+import CloudGraph from '@cloudgraph/sdk'
+import groupBy from 'lodash/groupBy'
+import { google } from '@google-cloud/dataproc/build/protos/protos'
+import gcpLoggerText from '../../properties/logger'
+import { GcpServiceInput } from '../../types'
+import { generateGcpErrorLog } from '../../utils'
+
+const lt = { ...gcpLoggerText }
+const { logger } = CloudGraph
+const serviceName = 'Dataproc Job'
+
+export interface RawGcpDataprocWorkflowTemplate extends Omit<google.cloud.dataproc.v1.IWorkflowTemplate, 'labels'> {
+  region: string
+  projectId: string
+  Labels: { [key: string]: string }
+}
+
+export default async ({
+  regions,
+  config,
+}: GcpServiceInput): Promise<{
+  [region: string]: RawGcpDataprocWorkflowTemplate[]
+}> => {
+  const workflowTemplateList: RawGcpDataprocWorkflowTemplate[] = []
+  const { projectId } = config
+  
+  for (const region of regions.split(',')) {
+    /**
+     * Get all the Dataproc Workflow Template
+     */
+    
+    try {
+        const dataprocClient = new WorkflowTemplateServiceClient({ 
+          ...config,
+          apiEndpoint: `${region}-dataproc.googleapis.com`,
+          projectId,
+        })
+
+        const iterable =  dataprocClient.listWorkflowTemplatesAsync({ 
+          parent: `projects/${projectId}/regions/${region}`
+        })
+        for await (const { labels, ...response } of iterable) {
+          if (response) {
+            workflowTemplateList.push({
+              ...response,
+              projectId,
+              region,
+              Labels: labels,
+            })
+          }
+        }
+      } catch (error) {
+        generateGcpErrorLog(serviceName, 'dataprocWorkflowTemplate:listWorkflowTemplatesAsync', error)
+      }
+    }
+
+    logger.debug(lt.foundResources(serviceName, workflowTemplateList.length))
+    return groupBy(workflowTemplateList, 'region')
+  }
+