<commit_before>/*
 * Copyright 2018 Spotify AB.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.spotify

import com.google.api.client.googleapis.auth.oauth2.GoogleCredential
import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport
import com.google.api.client.json.jackson2.JacksonFactory
import com.google.api.services.dataflow.Dataflow
import com.google.api.services.dataflow.model.{Job, JobMetrics}
import com.google.datastore.v1._
import com.google.datastore.v1.client.{Datastore, DatastoreHelper}
import com.spotify.scio._
import com.spotify.scio.runners.dataflow.DataflowResult
import org.apache.beam.sdk.PipelineResult.State
import org.joda.time.format.{ISODateTimeFormat, PeriodFormat}
import org.joda.time.{Instant, LocalDateTime, Seconds}
import shapeless.datatype.datastore.DatastoreType
import shapeless.datatype.datastore._

import scala.collection.JavaConverters._
import scala.language.higherKinds
import scala.util.{Failure, Success, Try}

/**
 * Shared functions to manage benchmark jobs and write metrics to DataStore
 *
 * This file is symlinked to scio-bench/src/main/scala/com/spotify/ScioBenchmark.scala so
 * that it can run with past Scio releases.
 */
object ScioBenchmarkSettings {
  val defaultProjectId: String = "data-integration-test"
  val numOfWorkers = 4
  def commonArgs(machineType: String = "n1-standard-4"): Array[String] = Array(
    "--runner=DataflowRunner",
    s"--numWorkers=$numOfWorkers",
    s"--workerMachineType=$machineType",
    "--autoscalingAlgorithm=NONE")

  val shuffleConf = Map("ShuffleService" -> Array("--experiments=shuffle_mode=service"))

  val circleCIEnv: Option[CircleCIEnv] = {
    val isCircleCIRun = sys.env.get("CIRCLECI").contains("true")

    if (isCircleCIRun) {
      (sys.env.get("CIRCLE_BUILD_NUM"), sys.env.get("CIRCLE_SHA1")) match {
        case (Some(buildNumber), Some(gitHash)) =>
          Some(CircleCIEnv(buildNumber.toLong, gitHash))
        case _ => throw new IllegalStateException("CIRCLECI env variable is set but not " +
          "CIRCLE_BUILD_NUM and CIRCLE_SHA1")
      }
    } else {
      PrettyPrint.print("CircleCI", "CIRCLECI env variable not found. Will not publish " +
        "benchmark results to Datastore.")
      None
    }
  }

  val BatchMetrics = Set("Elapsed", "TotalMemoryUsage", "TotalPdUsage",
    "TotalShuffleDataProcessed", "TotalSsdUsage", "TotalStreamingDataProcessed", "TotalVcpuTime")

  val StreamingMetrics = Set("CurrentMemoryUsage", "CurrentPdUsage", "CurrentVcpuCount",
    "TotalMemoryUsage", "TotalPdUsage", "TotalShuffleDataProcessed", "TotalSsdUsage",
    "TotalStreamingDataProcessed", "TotalVcpuTime")
}

final case class CircleCIEnv(buildNum: Long, gitHash: String)

object DataflowProvider {
  val dataflow: Dataflow = {
    val transport = GoogleNetHttpTransport.newTrustedTransport()
    val jackson = JacksonFactory.getDefaultInstance
    val credential = GoogleCredential.getApplicationDefault
    new Dataflow.Builder(transport, jackson, credential).build()
  }
}

trait BenchmarkLogger[F[_]] {
  def log(benchmarks: Iterable[BenchmarkResult]): F[Unit]
}

final case class ScioBenchmarkLogger[F[_]](loggers: BenchmarkLogger[F]*) {
  def log(benchmarks: BenchmarkResult*): Seq[F[Unit]] =
    loggers.map(_.log(benchmarks))
}

case class BenchmarkResult(
                            name: String,
                            elapsed: Option[Seconds],
                            startTime: LocalDateTime,
                            finishTime: Option[LocalDateTime],
                            state: State,
                            extraArgs: Array[String],
                            metrics: Map[String, String]
                          )<commit_msg>add Dataflow scopes to credential for StreamingBenchmark (#1338)<commit_after>/*
 * Copyright 2018 Spotify AB.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.spotify

import com.google.api.client.googleapis.auth.oauth2.GoogleCredential
import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport
import com.google.api.client.json.jackson2.JacksonFactory
import com.google.api.services.dataflow.{Dataflow, DataflowScopes}
import com.google.api.services.dataflow.model.{Job, JobMetrics}
import com.google.datastore.v1._
import com.google.datastore.v1.client.{Datastore, DatastoreHelper}
import com.spotify.scio._
import com.spotify.scio.runners.dataflow.DataflowResult
import org.apache.beam.sdk.PipelineResult.State
import org.joda.time.format.{ISODateTimeFormat, PeriodFormat}
import org.joda.time.{Instant, LocalDateTime, Seconds}
import shapeless.datatype.datastore.DatastoreType
import shapeless.datatype.datastore._

import scala.collection.JavaConverters._
import scala.language.higherKinds
import scala.util.{Failure, Success, Try}

/**
 * Shared functions to manage benchmark jobs and write metrics to DataStore
 *
 * This file is symlinked to scio-bench/src/main/scala/com/spotify/ScioBenchmark.scala so
 * that it can run with past Scio releases.
 */
object ScioBenchmarkSettings {
  val defaultProjectId: String = "data-integration-test"
  val numOfWorkers = 4
  def commonArgs(machineType: String = "n1-standard-4"): Array[String] = Array(
    "--runner=DataflowRunner",
    s"--numWorkers=$numOfWorkers",
    s"--workerMachineType=$machineType",
    "--autoscalingAlgorithm=NONE")

  val shuffleConf = Map("ShuffleService" -> Array("--experiments=shuffle_mode=service"))

  val circleCIEnv: Option[CircleCIEnv] = {
    val isCircleCIRun = sys.env.get("CIRCLECI").contains("true")

    if (isCircleCIRun) {
      (sys.env.get("CIRCLE_BUILD_NUM"), sys.env.get("CIRCLE_SHA1")) match {
        case (Some(buildNumber), Some(gitHash)) =>
          Some(CircleCIEnv(buildNumber.toLong, gitHash))
        case _ => throw new IllegalStateException("CIRCLECI env variable is set but not " +
          "CIRCLE_BUILD_NUM and CIRCLE_SHA1")
      }
    } else {
      PrettyPrint.print("CircleCI", "CIRCLECI env variable not found. Will not publish " +
        "benchmark results to Datastore.")
      None
    }
  }

  val BatchMetrics = Set("Elapsed", "TotalMemoryUsage", "TotalPdUsage",
    "TotalShuffleDataProcessed", "TotalSsdUsage", "TotalStreamingDataProcessed", "TotalVcpuTime")

  val StreamingMetrics = Set("CurrentMemoryUsage", "CurrentPdUsage", "CurrentVcpuCount",
    "TotalMemoryUsage", "TotalPdUsage", "TotalShuffleDataProcessed", "TotalSsdUsage",
    "TotalStreamingDataProcessed", "TotalVcpuTime")
}

final case class CircleCIEnv(buildNum: Long, gitHash: String)

object DataflowProvider {
  val dataflow: Dataflow = {
    val transport = GoogleNetHttpTransport.newTrustedTransport()
    val jackson = JacksonFactory.getDefaultInstance
    val credential = GoogleCredential.getApplicationDefault
      .createScoped(DataflowScopes.all())
    new Dataflow.Builder(transport, jackson, credential).build()
  }
}

trait BenchmarkLogger[F[_]] {
  def log(benchmarks: Iterable[BenchmarkResult]): F[Unit]
}

final case class ScioBenchmarkLogger[F[_]](loggers: BenchmarkLogger[F]*) {
  def log(benchmarks: BenchmarkResult*): Seq[F[Unit]] =
    loggers.map(_.log(benchmarks))
}

case class BenchmarkResult(
                            name: String,
                            elapsed: Option[Seconds],
                            startTime: LocalDateTime,
                            finishTime: Option[LocalDateTime],
                            state: State,
                            extraArgs: Array[String],
                            metrics: Map[String, String]
                          )
{'fraction_non_alphanumeric': 0.10382958910487719, 'fraction_numerical': 0.004423233616575486, 'mean_word_length': 3.6848418756815704, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2399171', 'n_tokens_mistral': 2590, 'n_tokens_neox': 2532, 'n_words': 605}