add support for etcd client authentication

common/scala/src/main/scala/org/apache/openwhisk/core/etcd/EtcdClient.scala

@@ -48,10 +48,16 @@ object RichListenableFuture {
 
 object EtcdClient {
   // hostAndPorts format: {HOST}:{PORT}[,{HOST}:{PORT},{HOST}:{PORT}, ...]
-  def apply(hostAndPorts: String)(implicit ece: ExecutionContextExecutor): EtcdClient = {
-    require(hostAndPorts != null)
-    val client: Client = Client.forEndpoints(hostAndPorts).withPlainText().build()
-    new EtcdClient(client)(ece)
+  def apply(config: EtcdConfig)(implicit ece: ExecutionContextExecutor): EtcdClient = {
+    require(config.hosts != null)
+    require(
+      (config.username.nonEmpty && config.password.nonEmpty) || (config.username.isEmpty && config.password.isEmpty))
+    val clientBuilder = Client.forEndpoints(config.hosts).withPlainText()
+    if (config.username.nonEmpty && config.password.nonEmpty) {
+      new EtcdClient(clientBuilder.withCredentials(config.username.get, config.password.get).build())
+    } else {
+      new EtcdClient(clientBuilder.build())(ece)
+    }
   }
 
   def apply(client: Client)(implicit ece: ExecutionContextExecutor): EtcdClient = {

common/scala/src/main/scala/org/apache/openwhisk/core/etcd/EtcdUtils.scala

@@ -30,7 +30,7 @@ import pureconfig.loadConfigOrThrow
 import scala.language.implicitConversions
 import scala.util.Try
 
-case class EtcdConfig(hosts: String)
+case class EtcdConfig(hosts: String, username: Option[String], password: Option[String])
 
 case class EtcdException(msg: String) extends Exception(msg)
 

core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/FPCPoolBalancer.scala

@@ -714,7 +714,7 @@ object FPCPoolBalancer extends LoadBalancerProvider {
       }
     }
 
-    val etcd = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd).hosts)
+    val etcd = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd))
 
     new FPCPoolBalancer(whiskConfig, instance, etcd, feedFactory)
   }

core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/FPCInvokerReactive.scala

@@ -82,7 +82,7 @@ class FPCInvokerReactive(config: WhiskConfig,
   private val logsProvider = SpiLoader.get[LogStoreProvider].instance(actorSystem)
   logging.info(this, s"LogStoreProvider: ${logsProvider.getClass}")
 
-  private val etcdClient = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd).hosts)
+  private val etcdClient = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd))
 
   private val grpcConfig = loadConfigOrThrow[GrpcServiceConfig](ConfigKeys.schedulerGrpcService)
 

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/Scheduler.scala

@@ -65,7 +65,7 @@ class Scheduler(schedulerId: SchedulerInstanceId, schedulerEndpoints: SchedulerE
   val producer = msgProvider.getProducer(config, Some(ActivationEntityLimit.MAX_ACTIVATION_LIMIT))
 
   val maxPeek = loadConfigOrThrow[Int](ConfigKeys.schedulerMaxPeek)
-  val etcdClient = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd).hosts)
+  val etcdClient = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd))
   val watcherService: ActorRef = actorSystem.actorOf(WatcherService.props(etcdClient))
   val leaseService =
     actorSystem.actorOf(LeaseKeepAliveService.props(etcdClient, schedulerId, watcherService))

tests/src/test/scala/org/apache/openwhisk/common/etcd/EtcdConfigTests.scala

@@ -0,0 +1,40 @@
+package org.apache.openwhisk.common.etcd
+
+import common.WskActorSystem
+import org.apache.openwhisk.core.etcd.{EtcdClient, EtcdConfig}
+import org.junit.runner.RunWith
+import org.scalatest.{FlatSpec, Matchers}
+import org.scalatest.junit.JUnitRunner
+
+import scala.concurrent.ExecutionContextExecutor
+
+
+@RunWith(classOf[JUnitRunner])
+class EtcdConfigTests
+  extends FlatSpec
+    with Matchers
+    with WskActorSystem {
+  behavior of "EtcdConfig"
+
+  implicit val ece: ExecutionContextExecutor = actorSystem.dispatcher
+
+  it should "create client when no auth is supplied through config" in {
+    val config = EtcdConfig("localhost:2379", None, None)
+
+    val client = EtcdClient(config)
+    client.close()
+  }
+
+  it should "create client when auth is supplied through config" in {
+    val config = EtcdConfig("localhost:2379", Some("username"), Some("password"))
+
+    val client = EtcdClient(config)
+    client.close()
+  }
+
+  it should "fail to create client when one of username or password is supplied in config" in {
+    val config = EtcdConfig("localhost:2379", None, Some("password"))
+
+    assertThrows[IllegalArgumentException](EtcdClient(config))
+  }
+}

tests/src/test/scala/org/apache/openwhisk/core/loadBalancer/test/FPCPoolBalancerTests.scala

@@ -63,7 +63,7 @@ class FPCPoolBalancerTests
 
   private implicit val transId = TransactionId.testing
   implicit val ece: ExecutionContextExecutor = actorSystem.dispatcher
-  private val etcd = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd).hosts)
+  private val etcd = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd))
 
   private val testInvocationNamespace = "test-invocation-namespace"
 

tests/src/test/scala/org/apache/openwhisk/core/scheduler/container/test/ContainerManagerTests.scala

@@ -1170,7 +1170,7 @@ class ContainerManager2Tests
     with DbUtils {
 
   implicit val dispatcher = actorSystem.dispatcher
-  val etcdClient = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd).hosts)
+  val etcdClient = EtcdClient(loadConfigOrThrow[EtcdConfig](ConfigKeys.etcd))
   val testInvocationNamespace = "test-invocation-namespace"
 
   override def afterAll(): Unit = {