Support for username / password credentials for Cassandra

docs/docs/configuration.md

@@ -17,6 +17,10 @@ If you installed chronos via package, run `/usr/bin/chronos run_jar --help`.
  points for Cassandra
 --cassandra_keyspace <arg> Keyspace to use for Cassandra
 --cassandra_port <arg> Port for Cassandra
+--cassandra_user <arg> User for Cassandra
+ (default = None)
+--cassandra_password <arg> Password for Cassandra
+ (default = None)
 --cassandra_stat_count_table <arg> Table to track stat counts in
  Cassandra
 --cassandra_table <arg> Table to use for Cassandra

src/main/scala/org/apache/mesos/chronos/scheduler/config/CassandraConfiguration.scala

@@ -12,6 +12,14 @@ trait CassandraConfiguration extends ScallopConf {
  descr = "Port for Cassandra",
  default = Some(9042))
 
+ lazy val cassandraUser = opt[String]("cassandra_user",
+ descr = "User",
+ default = None)
+
+ lazy val cassandraPassword = opt[String]("cassandra_password",
+ descr = "Password",
+ default = None)
+
  lazy val cassandraKeyspace = opt[String]("cassandra_keyspace",
  descr = "Keyspace to use for Cassandra",
  default = Some("metrics"))

src/main/scala/org/apache/mesos/chronos/scheduler/config/JobStatsModule.scala

@@ -16,14 +16,17 @@ class JobStatsModule(config: CassandraConfiguration) extends AbstractModule {
  def provideCassandraClusterBuilder(): Option[Cluster.Builder] = {
  config.cassandraContactPoints.get match {
  case Some(contactPoints) =>
- Some(
- Cluster.builder()
- .addContactPoints(contactPoints.split(","): _*)
- .withPort(config.cassandraPort())
- .withCompression(Compression.LZ4)
- .withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)
- .withLoadBalancingPolicy(LatencyAwarePolicy.builder(new RoundRobinPolicy).build)
- )
+ var builder = Cluster.builder()
+ .addContactPoints(contactPoints.split(","): _*)
+ .withPort(config.cassandraPort())
+ .withCompression(Compression.LZ4)
+ .withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)
+ .withLoadBalancingPolicy(LatencyAwarePolicy.builder(new RoundRobinPolicy).build)
+
+ if (config.cassandraUser.isDefined && config.cassandraPassword.isDefined) {
+ builder = builder.withCredentials(config.cassandraUser.get.orNull, config.cassandraPassword.get.orNull)
+ }
+ Some(builder)
  case _ =>
  None
  }

src/main/scala/org/apache/mesos/chronos/scheduler/jobs/stats/JobStats.scala

@@ -164,6 +164,9 @@ class JobStats @Inject()(clusterBuilder: Option[Cluster.Builder], config: Cassan
  case Some(c) =>
  try {
  val session = c.build.connect()
+ session.execute(new SimpleStatement(
+ s"CREATE KEYSPACE IF NOT EXISTS ${config.cassandraKeyspace()} WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };"
+ ))
  session.execute(new SimpleStatement(
  s"USE ${config.cassandraKeyspace()};"
  ))