SPARKC-651 move to Github Actions (#1312)

* SPARKC-651 fix BucketingRangeIndexSpec

* SPARKC-651 move to github actions

* SPARKC-651 use JAVA_HOME if CCM_JAVA_HOME is missing

* SPARKC-651 add waitForNode method with configurable timeout

ccm waits only for 30 seconds, the new method allows us
to have configurable timeout which may be useful  for
slower environments.

* SPARKC-651 add --skip-wait-other-notice

dse wait-other-notice causes the start process to hang for
10 minutes and eventually fail the build. Don't wait for
other nodes to notice the new coming node.

* SPARKC-651 add job matrix

2.5.x will be tested against C*/DSE versioned listed in main.yml

Author: jtgrabowski

.github/workflows/main.yml

@@ -0,0 +1,38 @@
+name: CI
+
+on:
+  # Triggers the workflow on push or pull request events but only for the master branch
+  push:
+    branches: [ b2.5, b3.0, master]
+  pull_request:
+    branches: [ b2.5, b3.0, master, SPARKC-651 ]
+
+  workflow_dispatch:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        scala:  [2.11.12]
+        db-version: [6.8.9, 3.11.9, 4.0-beta4]
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: ccm pip installation
+        uses: BSFishy/pip-action@v1
+        with:
+          packages: git+git://github.com/riptano/ccm.git@435f3210e16d0b648fbf33d6390d5ab4c9e630d4
+
+      - name: Setup Scala
+        uses: olafurpg/setup-scala@v10
+        with:
+          java-version: "[email protected]"
+
+      - name: sbt tests
+        env:
+          TEST_PARALLEL_TASKS: 1
+          CCM_CASSANDRA_VERSION: ${{ matrix.db-version }}
+          PUBLISH_VERSION: test
+        run: sbt/sbt ++${{ matrix.scala }} test it:test

.travis.yml

@@ -116,15 +116,16 @@ class BucketingRangeIndexSpec extends FlatSpec with ScalaCheckPropertyChecks wit
   private type LV = Long
 
   private val longTokenFactory = Murmur3TokenFactory
-
   private val LongFullRange = TokenRange[LV, LT](
       longTokenFactory.minToken,
       longTokenFactory.minToken,
       Set.empty,
       Murmur3TokenFactory)
 
+  "Murmur3Bucketing" should "map all tokens to a single wrapping range" in {
+    implicit val tokenOrdering: Ordering[LT] = longTokenFactory.tokenOrdering
+    implicit val tokenBucketing: MonotonicBucketing[LT] = longTokenFactory.tokenBucketing
 
-  "Murmur3Bucketing" should "  map all tokens to a single wrapping range" in {
     val singleRangeBucketing =
       new BucketingRangeIndex[LTR, LT](Seq(TokenRangeWithPartitionIndex(LongFullRange, 0)))
 
@@ -148,6 +149,9 @@ class BucketingRangeIndexSpec extends FlatSpec with ScalaCheckPropertyChecks wit
 
 
   "RandomBucketing" should " map all tokens to a single wrapping range" in {
+    implicit val tokenOrdering: Ordering[BT] = bigTokenFactory.tokenOrdering
+    implicit val tokenBucketing: MonotonicBucketing[BT] = bigTokenFactory.tokenBucketing
+
      val singleRangeBucketing =
       new BucketingRangeIndex[BITR, BT](Seq(TokenRangeWithPartitionIndex(bigFullRange, 0)))
 

connector/src/test/scala/com/datastax/spark/connector/rdd/partitioner/BucketingRangeIndexSpec.scala

@@ -10,7 +10,7 @@ import scala.sys.process._
 object Publishing extends sbt.librarymanagement.DependencyBuilders {
 
   val Version: String = {
-    sys.props.get("publish.version").getOrElse("git describe --tags" !!).stripLineEnd.stripPrefix("v")
+    sys.props.get("publish.version").orElse(sys.env.get("PUBLISH_VERSION")).getOrElse("git describe --tags" !!).stripLineEnd.stripPrefix("v")
   }
 
   val altReleaseDeploymentRepository = sys.props.get("publish.repository.name")

project/Publishing.scala

@@ -1,9 +1,10 @@
 import java.lang.management.ManagementFactory
-
 import com.sun.management.OperatingSystemMXBean
 import sbt.Tests._
 import sbt.{ForkOptions, TestDefinition}
 
+import scala.util.Try
+
 object Testing {
 
   private def interfacesImplementingFixture(c: Class[_], fixture: Class[_]): Seq[Class[_]] = {
@@ -42,16 +43,25 @@ object Testing {
       }
   }
 
+  /**
+    * Guesses if the given version is DSE or not.
+    * Internally it checks if the major version component is ge 6 which should suffice for a long time.
+    * CCM_IS_DSE env setting takes precedence over this guess.
+    */
+  private def isDse(version: Option[String]): Boolean = {
+    version.flatMap(v => Try(v.split('.').head.toInt >= 6).toOption).getOrElse(false)
+  }
+
   def getCCMJvmOptions = {
-    val ccmCassVersion = sys.env.get("CCM_CASSANDRA_VERSION").map(version => s"-Dccm.version=$version")
-    val ccmCassVersion2 = sys.env.get("CCM_CASSANDRA_VERSION").map(version => s"-Dcassandra.version=$version")
-    val dseInVersion = if (sys.env.get("CCM_CASSANDRA_VERSION").contains("dse")) Some(true) else None
-    val ccmDse = sys.env.get("CCM_IS_DSE").orElse(dseInVersion).map(isDSE => s"-Dccm.dse=$isDSE")
-    val ccmDse2 = sys.env.get("CCM_IS_DSE").orElse(dseInVersion).map(isDSE => s"-Ddse=$isDSE")
+    val dbVersion = sys.env.get("CCM_CASSANDRA_VERSION")
+    val ccmCassVersion = dbVersion.map(version => s"-Dccm.version=$version")
+    val ccmCassVersion2 = dbVersion.map(version => s"-Dcassandra.version=$version")
+    val ccmDse = sys.env.get("CCM_IS_DSE").map(_.toLowerCase == "true").orElse(Some(isDse(dbVersion)))
+      .map(isDSE => s"-Dccm.dse=$isDSE")
     val cassandraDirectory = sys.env.get("CCM_INSTALL_DIR").map(dir => s"-Dcassandra.directory=$dir")
-    val ccmJava = sys.env.get("CCM_JAVA_HOME").map(dir => s"-Dccm.java.home=$dir")
-    val ccmPath = sys.env.get("CCM_JAVA_HOME").map(dir => s"-Dccm.path=$dir/bin")
-    val options = Seq(ccmCassVersion, ccmDse, ccmCassVersion2, ccmDse2, cassandraDirectory, ccmJava, ccmPath)
+    val ccmJava = sys.env.get("CCM_JAVA_HOME").orElse(sys.env.get("JAVA_HOME")).map(dir => s"-Dccm.java.home=$dir")
+    val ccmPath = sys.env.get("CCM_JAVA_HOME").orElse(sys.env.get("JAVA_HOME")).map(dir => s"-Dccm.path=$dir/bin")
+    val options = Seq(ccmCassVersion, ccmDse, ccmCassVersion2, cassandraDirectory, ccmJava, ccmPath)
     options
   }
 

project/Testing.scala

@@ -22,7 +22,7 @@ case class CcmConfig(
     version: Version = Version.parse(System.getProperty("ccm.version", "3.11.6")),
     installDirectory: Option[String] = Option(System.getProperty("ccm.directory")),
     installBranch: Option[String] = Option(System.getProperty("ccm.branch")),
-    dseEnabled: Boolean = Option(System.getProperty("ccm.dse")).isDefined,
+    dseEnabled: Boolean = Option(System.getProperty("ccm.dse")).exists(_.toLowerCase == "true"),
     mode: ClusterMode = ClusterModes.fromEnvVar) {
 
   def withSsl(keystorePath: String, keystorePassword: String): CcmConfig = {

test-support/src/main/scala/com/datastax/spark/connector/ccm/CcmConfig.scala

@@ -1,10 +1,12 @@
 package com.datastax.spark.connector.ccm.mode
 
-import java.nio.file.Path
-
 import com.datastax.spark.connector.ccm.{CcmBridge, CcmConfig}
 import org.apache.commons.exec.CommandLine
 
+import java.nio.file.{Path, Paths}
+import scala.concurrent.duration.{Duration, DurationInt, FiniteDuration}
+import scala.util.control.NonFatal
+
 private[ccm] trait ClusterModeExecutor {
 
   protected val config: CcmConfig
@@ -32,4 +34,46 @@ private[ccm] trait ClusterModeExecutor {
     CcmBridge.execute(cli)
   }
 
+  def getLastRepositoryLogLines(linesCount: Int): Seq[String] = synchronized {
+    val log = Paths.get(
+      sys.props.get("user.home").get,
+      ".ccm",
+      "repository",
+      "ccm-repository.log").toString
+
+    getLastLogLines(log, linesCount)
+  }
+
+  def getLastLogLines(path: String, linesCount: Int): Seq[String] = synchronized {
+    val command = s"tail -$linesCount $path"
+    CcmBridge.execute(CommandLine.parse(command))
+  }
+
+  /**
+    * Waits for the node to become alive. The first check is performed after the first interval.
+    */
+  def waitForNode(nodeNo: Int, timeout: FiniteDuration, interval: Duration = 5.seconds): Boolean = {
+    val deadline = timeout.fromNow
+    while (!deadline.isOverdue()) {
+      Thread.sleep(interval.toMillis)
+      if (isAlive(nodeNo, interval)) {
+        return true
+      }
+    }
+    false;
+  }
+
+  private def isAlive(nodeNo: Int, timeout: Duration): Boolean = {
+    import java.net.Socket
+    val address = config.addressOfNode(nodeNo)
+    val socket = new Socket
+    try {
+      socket.connect(address, timeout.toMillis.toInt)
+      socket.close()
+      true
+    } catch {
+      case NonFatal(_) =>
+        false
+    }
+  }
 }

test-support/src/main/scala/com/datastax/spark/connector/ccm/mode/ClusterModeExecutor.scala

@@ -1,22 +1,53 @@
 package com.datastax.spark.connector.ccm.mode
 
-import java.io.File
+import java.io.{File, FileFilter}
 import java.nio.file.{Files, Path, Paths}
 import java.util.concurrent.atomic.AtomicBoolean
-
 import com.datastax.oss.driver.api.core.Version
 import com.datastax.spark.connector.ccm.CcmConfig
 import com.datastax.spark.connector.ccm.CcmConfig.V6_8_5
 import org.slf4j.{Logger, LoggerFactory}
 
+import scala.concurrent.duration.DurationInt
+import scala.util.Try
+import scala.util.control.NonFatal
+
 private[mode] trait DefaultExecutor extends ClusterModeExecutor {
   private val logger: Logger = LoggerFactory.getLogger(classOf[StandardModeExecutor])
 
   private val created = new AtomicBoolean()
 
+  private def waitForNode(nodeNo: Int): Unit = {
+    logger.info(s"Waiting for node $nodeNo to become alive...")
+    if (!waitForNode(nodeNo, 2.minutes)) {
+      throw new IllegalStateException(s"Timeouted on waiting for node $nodeNo")
+    }
+    logger.info(s"Node $nodeNo is alive")
+  }
+
+  private def logStdErr(nodeNo: Int): Unit = {
+    Try {
+      val linesCount = 1000
+      val logsDir = new File(s"${dir}/ccm_1/node${nodeNo}/logs/")
+
+      if (logsDir.exists() && logsDir.isDirectory) {
+        val stdErrFile = logsDir.listFiles().filter(_.getName.endsWith("stderr.log")).head
+        logger.error(s"Start command failed, here is the last $linesCount lines of startup-stderr file: \n" +
+          getLastLogLines(stdErrFile.getAbsolutePath, linesCount).mkString("\n"))
+      }
+    }
+  }
+
   override def start(nodeNo: Int): Unit = {
     val formattedJvmArgs = config.jvmArgs.map(arg => s" --jvm_arg=$arg").mkString(" ")
-    execute(s"node$nodeNo", "start", formattedJvmArgs + "--wait-for-binary-proto")
+    try {
+      execute(s"node$nodeNo", "start", formattedJvmArgs + "-v", "--skip-wait-other-notice")
+      waitForNode(nodeNo)
+    } catch {
+      case NonFatal(e) =>
+        logStdErr(nodeNo)
+        throw e
+    }
   }
 
   private def eventually[T](hint: String = "", f: =>T ): T = {
@@ -39,13 +70,15 @@ private[mode] trait DefaultExecutor extends ClusterModeExecutor {
   /**
     * Remove this once C* 4.0.0 is released.
     *
-    * This is a workaround that allows running it:test against 4.0.0-beta1. This version of C* is published as
-    * 4.0-beta1 which breaks versioning convention used in integration tests.
+    * This is a workaround that allows running it:test against 4.0.0-betaX and 4.0.0-rcX. These C* versions are
+    * published as 4.0-betaX and 4.0-rcX, lack of patch version breaks versioning convention used in integration tests.
     */
   private def adjustCassandraBetaVersion(version: String): String = {
     val beta = "4.0.0-beta(\\d+)".r
+    val rc = "4.0.0-rc(\\d+)".r
     version match {
       case beta(betaNo) => s"4.0-beta$betaNo"
+      case rc(rcNo) => s"4.0-rc$rcNo"
       case other => other
     }
   }
@@ -74,7 +107,14 @@ private[mode] trait DefaultExecutor extends ClusterModeExecutor {
         eventually(f = Files.exists(repositoryDir.resolve("bin")))
       }
 
-      execute( createArgs: _*)
+      try {
+        execute(createArgs: _*)
+      } catch {
+        case NonFatal(e) =>
+          Try(logger.error("Create command failed, here is the last 500 lines of ccm repository log: \n" +
+              getLastRepositoryLogLines(500).mkString("\n")))
+          throw e
+      }
 
       eventually("Checking to make sure repository was correctly expanded", {
         Files.exists(repositoryDir.resolve("bin"))