[CORE-40] Add cross-billing project spend report API

core/src/main/resources/swagger/api-docs.yaml

@@ -450,6 +450,55 @@ paths:
                 $ref: '#/components/schemas/ErrorReport'
         500:
           $ref: '#/components/responses/RawlsInternalError'
+  /api/billing/v2/spendReport:
+    get:
+      tags:
+        - billing_v2
+      summary: get spend report for all workspaces user has owner access to
+      description: get spend report for all workspaces user has owner access to
+      operationId: getSpendReportAllWorkspaces
+      parameters:
+        - name: startDate
+          in: query
+          description: start date of report (YYYY-MM-DD). Data included in report will start at 12 AM UTC on this date.
+          required: true
+          schema:
+            type: string
+            format: date
+        - name: endDate
+          in: query
+          description: end date of report (YYYY-MM-DD). Data included in report will end at 11:59 PM UTC on this date.
+          required: true
+          schema:
+            type: string
+            format: date
+      responses:
+        200:
+          description: Success
+          content:
+            'application/json':
+              schema:
+                $ref: '#/components/schemas/SpendReport'
+        400:
+          description: invalid spend report parameters
+          content:
+            'application/json':
+              schema:
+                $ref: '#/components/schemas/ErrorReport'
+        403:
+          description: You must be a project owner to view the spend report of a project
+          content:
+            'application/json':
+              schema:
+                $ref: '#/components/schemas/ErrorReport'
+        404:
+          description: The specified billing project could not be found
+          content:
+            'application/json':
+              schema:
+                $ref: '#/components/schemas/ErrorReport'
+        500:
+          $ref: '#/components/responses/RawlsInternalError'
   /api/billing/v2/{projectId}/spendReportConfiguration:
     get:
       tags:

core/src/main/scala/org/broadinstitute/dsde/rawls/Boot.scala

@@ -515,7 +515,8 @@ object Boot extends IOApp with LazyLogging {
           billingRepository,
           billingProfileManagerDAO,
           samDAO,
-          spendReportingServiceConfig
+          spendReportingServiceConfig,
+          workspaceServiceConstructor
         )
 
       val billingAdminServiceConstructor: RawlsRequestContext => BillingAdminService =

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/slick/RawlsBillingProjectComponent.scala

@@ -350,6 +350,16 @@ trait RawlsBillingProjectComponent {
         .result
         .map(_.headOption.map(RawlsBillingProjectRecord.toBillingProjectSpendExport))
 
+    def getBillingProjectsSpendConfiguration(
+      billingProjectNames: Seq[RawlsBillingProjectName]
+    ): ReadAction[Seq[Option[BillingProjectSpendExport]]] =
+      rawlsBillingProjectQuery
+        .withProjectNames(billingProjectNames)
+        .result
+        .map(projectRecords =>
+          projectRecords.map(record => Some(RawlsBillingProjectRecord.toBillingProjectSpendExport(record)))
+        )
+
     def insertOperations(operations: Seq[RawlsBillingProjectOperationRecord]): WriteAction[Unit] =
       (rawlsBillingProjectOperationQuery ++= operations).map(_ => ())
 

core/src/main/scala/org/broadinstitute/dsde/rawls/spendreporting/SpendReportingService.scala

@@ -17,15 +17,19 @@ import org.broadinstitute.dsde.rawls.dataaccess.{SamDAO, SlickDataSource}
 import org.broadinstitute.dsde.rawls.metrics.{GoogleInstrumented, HitRatioGauge, RawlsInstrumented}
 import org.broadinstitute.dsde.rawls.model.{SpendReportingAggregationKeyWithSub, _}
 import org.broadinstitute.dsde.rawls.spendreporting.SpendReportingService._
+import org.broadinstitute.dsde.rawls.workspace.WorkspaceService
 import org.broadinstitute.dsde.workbench.google2.GoogleBigQueryService
 import org.broadinstitute.dsde.rawls.{RawlsException, RawlsExceptionWithErrorReport}
 import org.broadinstitute.dsde.workbench.model.google.GoogleProject
 import org.joda.time.format.ISODateTimeFormat
 import org.joda.time.{DateTime, Days}
+import spray.json.JsArray
+import org.broadinstitute.dsde.rawls.model.WorkspaceJsonSupport.WorkspaceListResponseFormat
 
 import scala.concurrent.{ExecutionContext, Future}
 import scala.jdk.CollectionConverters._
 import scala.math.BigDecimal.RoundingMode
+import org.broadinstitute.dsde.rawls.model.WorkspaceAccessLevels.Owner
 
 object SpendReportingService {
   def constructor(
@@ -34,15 +38,18 @@ object SpendReportingService {
     billingRepository: BillingRepository,
     bpmDao: BillingProfileManagerDAO,
     samDAO: SamDAO,
-    spendReportingServiceConfig: SpendReportingServiceConfig
+    spendReportingServiceConfig: SpendReportingServiceConfig,
+    workspaceServiceConstructor: RawlsRequestContext => WorkspaceService
   )(ctx: RawlsRequestContext)(implicit executionContext: ExecutionContext): SpendReportingService =
-    new SpendReportingService(ctx,
-                              dataSource,
-                              bigQueryService,
-                              billingRepository: BillingRepository,
-                              bpmDao,
-                              samDAO,
-                              spendReportingServiceConfig
+    new SpendReportingService(
+      ctx,
+      dataSource,
+      bigQueryService,
+      billingRepository: BillingRepository,
+      bpmDao,
+      samDAO,
+      spendReportingServiceConfig,
+      workspaceServiceConstructor
     )
 
   val SpendReportingMetrics = "spendReporting"
@@ -133,6 +140,65 @@ object SpendReportingService {
     SpendReportingResults(aggregations.map(aggregateRows(allRows, _)).toList, summary)
   }
 
+  def extractCrossBillingProjectSpendReportingResults(
+    allRows: List[FieldValueList],
+    start: DateTime,
+    end: DateTime,
+    names: Map[GoogleProjectId, WorkspaceName]
+  ): List[SpendReportingResults] = {
+
+    val spendDetails = allRows.flatMap { row =>
+      val projectId = GoogleProjectId(row.get("project_id").getStringValue)
+      val workspaceName = names.getOrElse(
+        projectId,
+        throw RawlsExceptionWithErrorReport(
+          StatusCodes.InternalServerError,
+          s"unexpected project $projectId returned by BigQuery"
+        )
+      )
+      val totalCost = row.get("total_cost").getDoubleValue.toString
+      val computeCost = row.get("compute_cost").getDoubleValue.toString
+      val storageCost = row.get("storage_cost").getDoubleValue.toString
+      val currency = row.get("currency").getStringValue
+
+      // Each row gets a summary of compute, storage, and total
+      List(
+        SpendReportingForDateRange(
+          totalCost,
+          "0", // Ignoring credits for now; do we want to include them?
+          currency,
+          Option(start),
+          Option(end),
+          workspace = Some(workspaceName),
+          googleProjectId = Some(GoogleProject(projectId.value))
+        ),
+        SpendReportingForDateRange(
+          computeCost,
+          "0",
+          currency,
+          Option(start),
+          Option(end),
+          workspace = Some(workspaceName),
+          googleProjectId = Some(GoogleProject(projectId.value)),
+          category = Some(TerraSpendCategories.Compute)
+        ),
+        SpendReportingForDateRange(
+          storageCost,
+          "0",
+          currency,
+          Option(start),
+          Option(end),
+          workspace = Some(workspaceName),
+          googleProjectId = Some(GoogleProject(projectId.value)),
+          category = Some(TerraSpendCategories.Storage)
+        )
+      )
+    }
+
+    // TODO what format should the result be?  Do we need a new model?
+    spendDetails.map(details => SpendReportingResults(List.empty, details))
+  }
+
 }
 
 class SpendReportingService(
@@ -142,7 +208,8 @@ class SpendReportingService(
   billingRepository: BillingRepository,
   bpmDao: BillingProfileManagerDAO,
   samDAO: SamDAO,
-  spendReportingServiceConfig: SpendReportingServiceConfig
+  spendReportingServiceConfig: SpendReportingServiceConfig,
+  workspaceServiceConstructor: RawlsRequestContext => WorkspaceService
 )(implicit val executionContext: ExecutionContext)
     extends LazyLogging
     with RawlsInstrumented {
@@ -198,6 +265,20 @@ class SpendReportingService(
       )
     }
 
+  def getSpendExportConfigurations(projects: Seq[RawlsBillingProjectName]): Future[Seq[BillingProjectSpendExport]] =
+    dataSource
+      .inTransaction(_.rawlsBillingProjectQuery.getBillingProjectsSpendConfiguration(projects))
+      .recover { case _: RawlsException =>
+        throw RawlsExceptionWithErrorReport(
+          StatusCodes.BadRequest,
+          s"billing account not found on billing project" // TODO: identify problem
+        )
+      }
+      .map { exportOptions =>
+        exportOptions.collect { case Some(export) => export }
+
+      }
+
   def getWorkspaceGoogleProjects(projectName: RawlsBillingProjectName): Future[Map[GoogleProjectId, WorkspaceName]] =
     dataSource.inTransaction(_.workspaceQuery.listWithBillingProject(projectName)).map {
       _.collect {
@@ -222,11 +303,8 @@ class SpendReportingService(
     // all of which have optional subAggregationKeys and convert to Set[SpendReportingAggregationKey]
     val queryKeys = aggregations.flatMap(a => Set(Option(a.key), a.subAggregationKey).flatten)
     val tableName = config.spendExportTable.getOrElse(spendReportingServiceConfig.defaultTableName)
-    val timePartitionColumn: String = {
-      val isBroadTable = tableName == spendReportingServiceConfig.defaultTableName
-      // The Broad table uses a view with a different column name.
-      if (isBroadTable) spendReportingServiceConfig.defaultTimePartitionColumn else "_PARTITIONTIME"
-    }
+    val timePartitionColumn: String = getTimePartitionColumn(tableName)
+
     s"""
        | SELECT
        |  SUM(cost) as cost,
@@ -237,8 +315,70 @@ class SpendReportingService(
        | AND $timePartitionColumn BETWEEN @startDate AND @endDate
        | AND project.id in UNNEST(@projects)
        | GROUP BY currency ${queryKeys.map(_.bigQueryGroupByClause()).mkString}
-       |""".stripMargin
-      .replace("REPLACE_TIME_PARTITION_COLUMN", timePartitionColumn)
+       |""".stripMargin.replace("REPLACE_TIME_PARTITION_COLUMN", timePartitionColumn)
+  }
+
+  private def getTimePartitionColumn(tableName: String): String = {
+    val isBroadTable = tableName == spendReportingServiceConfig.defaultTableName
+    // The Broad table uses a view with a different column name.
+    if (isBroadTable) spendReportingServiceConfig.defaultTimePartitionColumn else "_PARTITIONTIME"
+  }
+
+  def getAllUserWorkspaceQuery(billingProjects: Map[BillingProjectSpendExport, Seq[GoogleProjectId]]): String = {
+    val baseQuery = s"""
+                       |  SELECT
+                       |    project.id AS project_id,
+                       |    project.name AS project_name,
+                       |    currency,
+                       |    CASE
+                       |      WHEN service.description IN ('Cloud Storage') THEN 'Storage'
+                       |      WHEN service.description IN ('Compute Engine', 'Google Kubernetes Engine') THEN 'Compute'
+                       |      ELSE 'Other'
+                       |    END AS spend_category,
+                       |    SUM(CAST(cost AS FLOAT64)) AS category_cost
+                       |  FROM
+                       |    _BILLING_ACCOUNT_TABLE
+                       |  where
+                       |    project.id in _PROJECT_ID_LIST AND
+                       |    _PARTITIONTIME BETWEEN @startDate AND @endDate
+                       |  GROUP BY
+                       |    project_id,
+                       |    project_name,
+                       |    currency,
+                       |    spend_category""".stripMargin.trim
+
+    val bpSubQuery = billingProjects
+      .map { bp =>
+        val tableName = bp._1.spendExportTable.getOrElse(spendReportingServiceConfig.defaultTableName)
+        val timePartitionColumn: String = getTimePartitionColumn(tableName)
+        baseQuery
+          .replace("_PARTITIONTIME", timePartitionColumn)
+          .replace("_BILLING_ACCOUNT_TABLE", tableName)
+          .replace("_PROJECT_ID_LIST", "(" + bp._2.map(id => s""""${id.value}"""").mkString(", ") + ")")
+      }
+      .mkString("\nUNION ALL\n")
+
+    s"""WITH spend_categories AS (
+       |$bpSubQuery
+       |)
+       |SELECT
+       |  project_id,
+       |  project_name,
+       |  SUM(category_cost) AS total_cost,
+       |  SUM(CASE WHEN spend_category = 'Storage' THEN category_cost ELSE 0 END) AS storage_cost,
+       |  SUM(CASE WHEN spend_category = 'Compute' THEN category_cost ELSE 0 END) AS compute_cost,
+       |  SUM(CASE WHEN spend_category = 'Other' THEN category_cost ELSE 0 END) AS other_cost,
+       |  currency
+       |FROM
+       |  spend_categories
+       |GROUP BY
+       |  project_id,
+       |  project_name,
+       |  currency
+       |ORDER BY
+       |  total_cost DESC
+       |limit 5
+       |""".stripMargin.trim
   }
 
   def setUpQuery(
@@ -270,6 +410,23 @@ class SpendReportingService(
     JobInfo.newBuilder(queryConfig).build()
   }
 
+  def setUpAllUserWorkspaceQuery(
+    query: String,
+    start: DateTime,
+    end: DateTime
+  ): JobInfo = {
+    def queryParam(value: String): QueryParameterValue =
+      QueryParameterValue.newBuilder().setType(StandardSQLTypeName.STRING).setValue(value).build()
+
+    val queryConfig = QueryJobConfiguration
+      .newBuilder(query)
+      .addNamedParameter("startDate", queryParam(toISODateString(start)))
+      .addNamedParameter("endDate", queryParam(toISODateString(end)))
+      .build()
+
+    JobInfo.newBuilder(queryConfig).build()
+  }
+
   def logSpendQueryStats(stats: JobStatistics.QueryStatistics): Unit = {
     if (stats.getCacheHit) cacheHitRate().hit() else cacheHitRate().miss()
     bytesProcessedCounter += stats.getEstimatedBytesProcessed
@@ -359,4 +516,58 @@ class SpendReportingService(
         case ex: Exception =>
           Future.failed(RawlsExceptionWithErrorReport(ErrorReport(StatusCodes.InternalServerError, ex)))
       }
+
+  def getSpendForAllWorkspaces(
+    start: DateTime,
+    end: DateTime
+  ): Future[List[SpendReportingResults]] = {
+    validateReportParameters(start, end)
+    for {
+      workspaces <- getOwnerWorkspaces()
+      projectNames = workspaces
+        .map(wsResp =>
+          wsResp.workspace.googleProject -> WorkspaceName(wsResp.workspace.namespace, wsResp.workspace.name)
+        )
+        .toMap // Map[GoogleProjectId, WorkspaceName]
+      billing <- getBillingSpendExportsForWorkspaces(workspaces)
+      query = getAllUserWorkspaceQuery(billing)
+      queryJob = setUpAllUserWorkspaceQuery(query, start, end)
+
+      job: Job <- bigQueryService.use(_.runJob(queryJob)).unsafeToFuture().map(_.waitFor())
+      _ = logSpendQueryStats(job.getStatistics[JobStatistics.QueryStatistics])
+      result = job.getQueryResults()
+    } yield result.getValues.asScala.toList match {
+      case Nil =>
+        throw RawlsExceptionWithErrorReport(
+          StatusCodes.NotFound,
+          s"no spend data found between dates ${toISODateString(start)} and ${toISODateString(end)}"
+        ) // TODO update this
+      case rows => extractCrossBillingProjectSpendReportingResults(rows, start, end, projectNames)
+    }
+  }
+
+  def getOwnerWorkspaces(): Future[Seq[WorkspaceListResponse]] =
+    workspaceServiceConstructor(ctx).listWorkspaces(WorkspaceFieldSpecs(), -1) map {
+      case JsArray(jsArray) =>
+        val workspaces = jsArray.map(_.convertTo[WorkspaceListResponse])
+        workspaces.filter(_.accessLevel == Owner)
+      case _ => throw new IllegalArgumentException("Expected a JsArray")
+    }
+
+  def getBillingSpendExportsForWorkspaces(
+    workspaces: Seq[WorkspaceListResponse]
+  ): Future[Map[BillingProjectSpendExport, Seq[GoogleProjectId]]] = {
+    val groupedWorkspaces = workspaces.groupBy(_.workspace.namespace)
+    val billingProjects = groupedWorkspaces.keys.map(RawlsBillingProjectName).toList
+//    billingProjects.map(project =>
+//      requireProjectAction(project, SamBillingProjectActions.readSpendReport)
+//    ) // TODO a better way to handle this?
+
+    getSpendExportConfigurations(billingProjects).map { exportConfigs =>
+      exportConfigs.map { config =>
+        config -> groupedWorkspaces(config.billingProjectName.value).map(ws => ws.workspace.googleProject)
+      }.toMap
+    }
+  }
+
 }