feat(dataproc): add dataproc job

Author: james-zhou-inspire11

src/enums/schemasMap.ts

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

src/enums/serviceMap.ts

@@ -38,6 +38,7 @@ import GcpApiKey from '../services/apiKey'
 import GcpComputeProject from '../services/computeProject'
 import GcpDataprocCluster from '../services/dataprocCluster'
 import GcpDataprocAutoscalingPolicy from '../services/dataprocAutoscalingPolicy'
+import GcpDataprocJob from '../services/dataprocJob'
 
 /**
  * serviceMap is an object that contains all currently supported services
@@ -82,5 +83,6 @@ export default {
   [services.computeProject]: GcpComputeProject,
   [services.dataprocCluster]: GcpDataprocCluster,
   [services.dataprocAutoscalingPolicy]: GcpDataprocAutoscalingPolicy,
+  [services.dataprocJob]: GcpDataprocJob,
   tag: GcpTag,
 }

src/enums/services.ts

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

src/services/dataprocJob/data.ts

@@ -0,0 +1,60 @@
+import { JobControllerClient } 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 RawGcpDataprocJob extends Omit<google.cloud.dataproc.v1.IJob, 'jobUuid'|'labels'> {
+  id: string
+  region: string
+  projectId: string
+  Labels: { [key: string]: string }
+}
+
+export default async ({
+  regions,
+  config,
+}: GcpServiceInput): Promise<{
+  [region: string]: RawGcpDataprocJob[]
+}> => {
+    const jobList: RawGcpDataprocJob[] = []
+    const { projectId } = config
+
+    for (const region of regions.split(',')) {
+      /**
+       * Get all the Dataproc Job
+       */
+        
+      try {
+        const dataprocClient = new JobControllerClient({ 
+          ...config,
+          apiEndpoint: `${region}-dataproc.googleapis.com`,
+          projectId,
+        })
+      
+        const iterable =  dataprocClient.listJobsAsync({ projectId, region })
+        for await (const { jobUuid, labels, ...response } of iterable) {
+          if (response) {
+            jobList.push({
+              ...response,
+              id: jobUuid,
+              projectId,
+              region,
+              Labels: labels,
+            })
+          }
+        }
+      } catch (error) {
+        generateGcpErrorLog(serviceName, 'dataprocJob:listJobsAsync', error)
+      }
+    }
+
+    logger.debug(lt.foundResources(serviceName, jobList.length))
+    return groupBy(jobList, 'region')
+  }

src/services/dataprocJob/format.ts

@@ -0,0 +1,275 @@
+import cuid from 'cuid'
+import { google } from '@google-cloud/dataproc/build/protos/protos'
+import {
+  GcpDataprocHadoopJob,
+  GcpDataprocHiveJob,
+  GcpDataprocJob,
+  GcpDataprocJobStatus,
+  GcpDataprocJobYarnApplication,
+  GcpDataprocPigJob,
+  GcpDataprocPrestoJob,
+  GcpDataprocPysparkJob,
+  GcpDataprocSparkJob,
+  GcpDataprocspArkRJob,
+  GcpDataprocSparkSqlJob,
+  GcpKeyValue,
+} from '../../types/generated'
+import { RawGcpDataprocJob } from './data'
+import { toISOString } from '../../utils/dateutils'
+import { enumKeyToString, formatKeyValueMap, formatLabelsFromMap } from '../../utils/format'
+
+const formatLoggingConfig = (loggingConfig: { [k: string]: google.cloud.dataproc.v1.LoggingConfig.Level }): GcpKeyValue[] => {
+  return Object.keys(loggingConfig || {}).map(key => ({
+    id: cuid(),
+    key,
+    value: enumKeyToString(google.cloud.dataproc.v1.LoggingConfig.Level, loggingConfig[key]),
+  }))
+}
+
+const formatStatus = ({
+  state,
+  details,
+  stateStartTime,
+  substate,
+}: google.cloud.dataproc.v1.IJobStatus): GcpDataprocJobStatus => {
+  return {
+    id: cuid(),
+    state: enumKeyToString(google.cloud.dataproc.v1.JobStatus.State, state),
+    details,
+    stateStartTime: toISOString(stateStartTime?.seconds?.toString()) || '',
+    substate: enumKeyToString(google.cloud.dataproc.v1.JobStatus.Substate, substate),
+  }
+}
+
+const formatYarnApplication = ({
+  name,
+  state,
+  progress,
+  trackingUrl,
+}: google.cloud.dataproc.v1.IYarnApplication): GcpDataprocJobYarnApplication => {
+  return {
+    id: cuid(),
+    name,
+    state: enumKeyToString(google.cloud.dataproc.v1.YarnApplication.State, state),
+    progress,
+    trackingUrl,
+  }
+}
+
+const formatHadoopJob = ({
+  mainJarFileUri,
+  mainClass,
+  args,
+  jarFileUris,
+  fileUris,
+  archiveUris,
+  properties = {},
+  loggingConfig = {},
+}: google.cloud.dataproc.v1.IHadoopJob): GcpDataprocHadoopJob => {
+  return {
+    mainJarFileUri,
+    mainClass,
+    args,
+    jarFileUris,
+    fileUris,
+    archiveUris,
+    properties: formatKeyValueMap(properties),
+    loggingConfig: formatLoggingConfig(loggingConfig?.driverLogLevels || {}),
+  }
+}
+
+const formatSparkJob = ({
+  mainJarFileUri,
+  mainClass,
+  args,
+  jarFileUris,
+  fileUris,
+  archiveUris,
+  properties = {},
+  loggingConfig = {},
+}: google.cloud.dataproc.v1.ISparkJob): GcpDataprocSparkJob => {
+  return {
+    mainJarFileUri,
+    mainClass,
+    args,
+    jarFileUris,
+    fileUris,
+    archiveUris,
+    properties: formatKeyValueMap(properties || {}),
+    loggingConfig: formatLoggingConfig(loggingConfig?.driverLogLevels || {}),
+  }
+}
+
+const formatPysparkJob = ({
+  mainPythonFileUri,
+  args = [],
+  pythonFileUris = [],
+  jarFileUris = [],
+  fileUris = [],
+  archiveUris = [],
+  properties = {},
+  loggingConfig = {},
+}: google.cloud.dataproc.v1.IPySparkJob): GcpDataprocPysparkJob => {
+  return {
+    mainPythonFileUri,
+    args,
+    pythonFileUris,
+    jarFileUris,
+    fileUris,
+    archiveUris,
+    properties: formatKeyValueMap(properties || {}),
+    loggingConfig: formatLoggingConfig(loggingConfig?.driverLogLevels || {}),
+  }
+}
+
+const formatHiveJob = ({
+  queryFileUri,
+  queryList = {},
+  continueOnFailure = false,
+  scriptVariables = {},
+  properties = {},
+  jarFileUris = [],
+}: google.cloud.dataproc.v1.IHiveJob): GcpDataprocHiveJob => {
+  return {
+    queryFileUri,
+    queryList: queryList?.queries || [],
+    continueOnFailure,
+    scriptVariables: formatKeyValueMap(scriptVariables || {}),
+    properties: formatKeyValueMap(properties || {}),
+    jarFileUris,
+  }
+}
+
+const formatPigJob = ({
+  queryFileUri,
+  queryList = {},
+  continueOnFailure = false,
+  scriptVariables = {},
+  properties = {},
+  jarFileUris = [],
+  loggingConfig = {},
+}: google.cloud.dataproc.v1.IPigJob): GcpDataprocPigJob => {
+  return {
+    queryFileUri,
+    queryList: queryList?.queries || [],
+    continueOnFailure,
+    scriptVariables: formatKeyValueMap(scriptVariables || {}),
+    properties: formatKeyValueMap(properties || {}),
+    jarFileUris,
+    loggingConfig: formatLoggingConfig(loggingConfig?.driverLogLevels || {}),
+  }
+}
+
+const formatSparkRJob = ({
+  mainRFileUri,
+  args = [],
+  fileUris = [],
+  archiveUris = [],
+  properties = {},
+  loggingConfig = {},
+}: google.cloud.dataproc.v1.ISparkRJob): GcpDataprocspArkRJob => {
+  return {
+    mainRFileUri,
+    args,
+    fileUris,
+    archiveUris,
+    properties: formatKeyValueMap(properties || {}),
+    loggingConfig: formatLoggingConfig(loggingConfig?.driverLogLevels || {}),
+  }
+}
+
+const formatSparkSqlJob = ({
+  queryFileUri,
+  queryList = {},
+  scriptVariables = {},
+  properties = {},
+  jarFileUris = [],
+  loggingConfig = {},
+}: google.cloud.dataproc.v1.ISparkSqlJob): GcpDataprocSparkSqlJob => {
+  return {
+    queryFileUri,
+    queryList: queryList?.queries || [],
+    scriptVariables: formatKeyValueMap(scriptVariables || {}),
+    properties: formatKeyValueMap(properties || {}),
+    jarFileUris: jarFileUris || [],
+    loggingConfig: formatLoggingConfig(loggingConfig?.driverLogLevels || {}),
+  }
+}
+
+const formatPrestoJob = ({
+  queryFileUri,
+  queryList = {},
+  continueOnFailure = false,
+  outputFormat,
+  clientTags = [],
+  properties = {},
+  loggingConfig = {},
+}: google.cloud.dataproc.v1.IPrestoJob): GcpDataprocPrestoJob => {
+  return {
+    queryFileUri,
+    queryList: queryList?.queries || [],
+    continueOnFailure: continueOnFailure || false,
+    outputFormat,
+    clientTags,
+    properties: formatKeyValueMap(properties || {}),
+    loggingConfig: formatLoggingConfig(loggingConfig?.driverLogLevels || {}),
+  }
+}
+
+export default ({
+  service,
+  region,
+}: {
+  service: RawGcpDataprocJob
+  region: string
+}): GcpDataprocJob => {
+  const {
+    id,
+    projectId,
+    reference,
+    placement,
+    hadoopJob = {},
+    sparkJob = {},
+    pysparkJob = {},
+    hiveJob = {},
+    pigJob = {},
+    sparkRJob = {},
+    sparkSqlJob = {},
+    prestoJob = {},
+    status = {},
+    statusHistory = [],
+    yarnApplications = [],
+    driverOutputResourceUri,
+    driverControlFilesUri,
+    scheduling = {},
+    done,
+    Labels = {},
+  } = service
+
+  return {
+    id,
+    projectId,
+    region,
+    name: reference?.jobId || '',
+    placementClusterName: placement?.clusterName || '',
+    placementClusterUuid: placement?.clusterUuid || '',
+    placementClusterLabels: formatKeyValueMap(placement?.clusterLabels || {}),
+    hadoopJob: formatHadoopJob(hadoopJob),
+    sparkJob: formatSparkJob(sparkJob),
+    pysparkJob: formatPysparkJob(pysparkJob),
+    hiveJob: formatHiveJob( hiveJob),
+    pigJob: formatPigJob(pigJob),
+    sparkRJob: formatSparkRJob(sparkRJob),
+    sparkSqlJob: formatSparkSqlJob(sparkSqlJob),
+    prestoJob: formatPrestoJob(prestoJob),
+    status: formatStatus(status),
+    statusHistory: statusHistory?.map(formatStatus),
+    yarnApplications: yarnApplications?.map(formatYarnApplication),
+    driverOutputResourceUri,
+    driverControlFilesUri,
+    schedulingMaxFailuresPerHour: scheduling?.maxFailuresPerHour || 0,
+    schedulingMaxFailuresTotal: scheduling?.maxFailuresTotal || 0,
+    done,
+    labels: formatLabelsFromMap(Labels),
+  }
+}

src/services/dataprocJob/index.ts

@@ -0,0 +1,13 @@
+import {Service} from '@cloudgraph/sdk'
+import BaseService from '../base'
+import format from './format'
+import getData from './data'
+import mutation from './mutation'
+
+export default class GcpDataprocCluster extends BaseService implements Service {
+  format = format.bind(this)
+
+  getData = getData.bind(this)
+
+  mutation = mutation
+}

src/services/dataprocJob/mutation.ts

@@ -0,0 +1,5 @@
+export default `mutation($input: [AddgcpDataprocJobInput!]!) {
+  addgcpDataprocJob(input: $input, upsert: true) {
+    numUids
+  }
+}`;