- removed usage implicit custom ActorMaterializer

- upgraded to Akka Http 10.2.3
- changed the way how Https set up when host name verification is disabled
- changed RestartSink.withBackoff to accept RestartSettings
- changed akka.actor.FSM.setTimer => akka.actor.FSM.startTimerAtFixedRate
- changed akka.actor.Scheduler.schedule => akka.actor.Scheduler.scheduleAtFixedRate
- updated to the new signature of Source.actorRef
- replaced deprecated HttpResponse.copy
- replaced Gzip with Coders.Gzip
- fixed the scalafmt in tests
- removed the implicit Materializer shutdown considering its lifecycle is managed by akka
- increased the ansible wait time
- switched to classic networking
- disabled JFR

Author: vrann

ansible/roles/controller/tasks/deploy.yml

@@ -361,4 +361,4 @@
   register: result
   until: result.status == 200
   retries: 12
-  delay: 5
+  delay: 10

ansible/roles/controller/tasks/join_akka_cluster.yml

@@ -32,7 +32,7 @@
     env: >-
       {{ env | combine({
         'CONFIG_akka_cluster_seedNodes_' ~ seedNode.0:
-          'akka.tcp://controller-actor-system@'~seedNode.1~':'~(controller.akka.cluster.basePort+seedNode.0)
+          'akka://controller-actor-system@'~seedNode.1~':'~(controller.akka.cluster.basePort+seedNode.0)
       }) }}
   with_indexed_items: "{{ controller.akka.cluster.seedNodes }}"
   loop_control:
@@ -42,11 +42,11 @@
   vars:
     akka_env:
       "CONFIG_akka_actor_provider": "{{ controller.akka.provider }}"
-      "CONFIG_akka_remote_netty_tcp_hostname":
+      "CONFIG_akka_remote_artery_canonical_hostname":
         "{{ controller.akka.cluster.host[(controller_index | int)] }}"
-      "CONFIG_akka_remote_netty_tcp_port":
+      "CONFIG_akka_remote_artery_canonical_port":
         "{{ controller.akka.cluster.basePort + (controller_index | int) }}"
-      "CONFIG_akka_remote_netty_tcp_bindPort":
+      "CONFIG_akka_remote_artery_bind_port":
         "{{ controller.akka.cluster.bindPort }}"
   set_fact:
     env: "{{ env | combine(akka_env) }}"

common/scala/src/main/resources/application.conf

@@ -18,6 +18,10 @@
 # default application configuration file for akka
 include "logging"
 
+akka {
+    java-flight-recorder.enabled = false
+}
+
 akka.http {
     client {
         parsing.illegal-header-warnings = off

common/scala/src/main/scala/org/apache/openwhisk/common/Https.scala

@@ -20,10 +20,8 @@ package org.apache.openwhisk.common
 import java.io.{FileInputStream, InputStream}
 import java.security.{KeyStore, SecureRandom}
 import javax.net.ssl.{KeyManagerFactory, SSLContext, TrustManagerFactory}
-
-import akka.http.scaladsl.ConnectionContext
+import akka.http.scaladsl.{ConnectionContext, HttpsConnectionContext}
 import akka.stream.TLSClientAuth
-import com.typesafe.sslconfig.akka.AkkaSSLConfig
 
 object Https {
   case class HttpsConfig(keystorePassword: String, keystoreFlavor: String, keystorePath: String, clientAuth: String)
@@ -35,8 +33,16 @@ object Https {
     cs
   }
 
-  def connectionContext(httpsConfig: HttpsConfig, sslConfig: Option[AkkaSSLConfig] = None) = {
+  def httpsInsecureClient(context: SSLContext): HttpsConnectionContext =
+    ConnectionContext.httpsClient((host, port) => {
+      val engine = context.createSSLEngine(host, port)
+      engine.setUseClientMode(true)
+      // WARNING: this creates an SSL Engine without enabling endpoint identification/verification procedures
+      // Disabling host name verification is a very bad idea, please don't unless you have a very good reason to.
+      engine
+    })
 
+  def applyHttpsConfig(httpsConfig: HttpsConfig, withDisableHostnameVerification: Boolean = false): SSLContext = {
     val keyFactoryType = "SunX509"
     val clientAuth = {
       if (httpsConfig.clientAuth.toBoolean)
@@ -63,7 +69,27 @@ object Https {
 
     val sslContext: SSLContext = SSLContext.getInstance("TLS")
     sslContext.init(keyManagerFactory.getKeyManagers, trustManagerFactory.getTrustManagers, new SecureRandom)
+    sslContext
+  }
+
+  def connectionContextClient(httpsConfig: HttpsConfig,
+                              withDisableHostnameVerification: Boolean = false): HttpsConnectionContext = {
+    val sslContext = applyHttpsConfig(httpsConfig, withDisableHostnameVerification)
+    connectionContextClient(sslContext, withDisableHostnameVerification)
+  }
+
+  def connectionContextClient(sslContext: SSLContext,
+                              withDisableHostnameVerification: Boolean): HttpsConnectionContext = {
+    if (withDisableHostnameVerification) {
+      httpsInsecureClient(sslContext)
+    } else {
+      ConnectionContext.httpsClient(sslContext)
+    }
+  }
 
-    ConnectionContext.https(sslContext, sslConfig, clientAuth = clientAuth)
+  def connectionContextServer(httpsConfig: HttpsConfig,
+                              withDisableHostnameVerification: Boolean = false): HttpsConnectionContext = {
+    val sslContext: SSLContext = applyHttpsConfig(httpsConfig, withDisableHostnameVerification)
+    ConnectionContext.httpsServer(sslContext)
   }
 }

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationFileLogStore.scala

@@ -32,7 +32,7 @@ import java.time.Instant
 import akka.NotUsed
 import akka.actor.ActorSystem
 import akka.stream.alpakka.file.scaladsl.LogRotatorSink
-import akka.stream.{Graph, SinkShape, UniformFanOutShape}
+import akka.stream.{Graph, RestartSettings, SinkShape, UniformFanOutShape}
 import akka.stream.scaladsl.{Broadcast, Flow, GraphDSL, Keep, MergeHub, RestartSink, Sink, Source}
 import akka.util.ByteString
 
@@ -87,32 +87,33 @@ class DockerToActivationFileLogStore(system: ActorSystem, destinationDirectory:
   protected val writeToFile: Sink[ByteString, _] = MergeHub
     .source[ByteString]
     .batchWeighted(bufferSize.toBytes, _.length, identity)(_ ++ _)
-    .to(RestartSink.withBackoff(minBackoff = 1.seconds, maxBackoff = 60.seconds, randomFactor = 0.2) { () =>
-      LogRotatorSink(() => {
-        val maxSize = bufferSize.toBytes
-        var bytesRead = maxSize
-        element =>
-          {
-            val size = element.size
-            if (bytesRead + size > maxSize) {
-              bytesRead = size
-              val logFilePath = destinationDirectory.resolve(s"userlogs-${Instant.now.toEpochMilli}.log")
-              logging.info(this, s"Rotating log file to '$logFilePath'")
-              try {
-                Files.createFile(logFilePath)
-                Files.setPosixFilePermissions(logFilePath, perms)
-              } catch {
-                case t: Throwable =>
-                  logging.error(this, s"Couldn't create userlogs file: $t")
-                  throw t
+    .to(RestartSink.withBackoff(RestartSettings(minBackoff = 1.seconds, maxBackoff = 60.seconds, randomFactor = 0.2)) {
+      () =>
+        LogRotatorSink(() => {
+          val maxSize = bufferSize.toBytes
+          var bytesRead = maxSize
+          element =>
+            {
+              val size = element.size
+              if (bytesRead + size > maxSize) {
+                bytesRead = size
+                val logFilePath = destinationDirectory.resolve(s"userlogs-${Instant.now.toEpochMilli}.log")
+                logging.info(this, s"Rotating log file to '$logFilePath'")
+                try {
+                  Files.createFile(logFilePath)
+                  Files.setPosixFilePermissions(logFilePath, perms)
+                } catch {
+                  case t: Throwable =>
+                    logging.error(this, s"Couldn't create userlogs file: $t")
+                    throw t
+                }
+                Some(logFilePath)
+              } else {
+                bytesRead += size
+                None
               }
-              Some(logFilePath)
-            } else {
-              bytesRead += size
-              None
             }
-          }
-      })
+        })
     })
     .run()
 

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/DockerToActivationLogStore.scala

@@ -21,7 +21,6 @@ import java.time.Instant
 
 import akka.NotUsed
 import akka.actor.ActorSystem
-import akka.stream.ActorMaterializer
 import akka.stream.scaladsl.Sink
 import akka.stream.scaladsl.Flow
 import akka.stream.scaladsl.Source
@@ -62,7 +61,7 @@ object DockerToActivationLogStore {
  */
 class DockerToActivationLogStore(system: ActorSystem) extends LogStore {
   implicit val ec: ExecutionContext = system.dispatcher
-  implicit val mat: ActorMaterializer = ActorMaterializer()(system)
+  implicit val actorSystem: ActorSystem = system
 
   /* "json-file" is the log-driver that writes out to file */
   override val containerParameters = Map("--log-driver" -> Set("json-file"))

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/ElasticSearchLogStore.scala

@@ -75,8 +75,6 @@ class ElasticSearchLogStore(
         elasticSearchConfig.logSchema.time)
   }
 
-  implicit val actorSystem = system
-
   private val esClient = new ElasticSearchRestClient(
     elasticSearchConfig.protocol,
     elasticSearchConfig.host,

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/SplunkLogStore.scala

@@ -21,6 +21,7 @@ import java.time.Instant
 import java.time.temporal.ChronoUnit
 
 import akka.actor.ActorSystem
+import akka.http.scaladsl.ConnectionContext
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.client.RequestBuilding.Post
 import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
@@ -32,14 +33,13 @@ import akka.http.scaladsl.model.Uri.Path
 import akka.http.scaladsl.model.headers.Authorization
 import akka.http.scaladsl.model.headers.BasicHttpCredentials
 import akka.http.scaladsl.unmarshalling.Unmarshal
-import akka.stream.ActorMaterializer
 import akka.stream.OverflowStrategy
 import akka.stream.QueueOfferResult
 import akka.stream.scaladsl.Flow
 import akka.stream.scaladsl.Keep
 import akka.stream.scaladsl.Sink
 import akka.stream.scaladsl.Source
-import com.typesafe.sslconfig.akka.AkkaSSLConfig
+import javax.net.ssl.{SSLContext, SSLEngine}
 import pureconfig._
 import pureconfig.generic.auto._
 
@@ -89,7 +89,6 @@ class SplunkLogStore(
     extends LogDriverLogStore(actorSystem) {
   implicit val as = actorSystem
   implicit val ec = as.dispatcher
-  implicit val materializer = ActorMaterializer()
   private val logging = new AkkaLogging(actorSystem.log)
 
   private val splunkApi = Path / "services" / "search" / "jobs" //see http://docs.splunk.com/Documentation/Splunk/6.6.3/RESTREF/RESTsearch#search.2Fjobs
@@ -98,12 +97,22 @@ class SplunkLogStore(
 
   val maxPendingRequests = 500
 
+  def createInsecureSslEngine(host: String, port: Int): SSLEngine = {
+    val engine = SSLContext.getDefault.createSSLEngine(host, port)
+    engine.setUseClientMode(true)
+
+    // WARNING: this creates an SSL Engine without enabling endpoint identification/verification procedures
+    // Disabling host name verification is a very bad idea, please don't unless you have a very good reason to.
+
+    engine
+  }
+
   val defaultHttpFlow = Http().cachedHostConnectionPoolHttps[Promise[HttpResponse]](
     host = splunkConfig.host,
     port = splunkConfig.port,
     connectionContext =
       if (splunkConfig.disableSNI)
-        Http().createClientHttpsContext(AkkaSSLConfig().mapSettings(s => s.withLoose(s.loose.withDisableSNI(true))))
+        ConnectionContext.httpsClient(createInsecureSslEngine _)
       else Http().defaultClientHttpsContext)
 
   override def fetchLogs(namespace: String,

common/scala/src/main/scala/org/apache/openwhisk/core/database/ActivationFileStorage.scala

@@ -22,7 +22,8 @@ import java.nio.file.{Files, Path}
 import java.time.Instant
 import java.util.EnumSet
 
-import akka.stream.ActorMaterializer
+import akka.actor.ActorSystem
+import akka.stream.RestartSettings
 import akka.stream.alpakka.file.scaladsl.LogRotatorSink
 import akka.stream.scaladsl.{Flow, MergeHub, RestartSink, Sink, Source}
 import akka.util.ByteString
@@ -37,38 +38,38 @@ import scala.concurrent.duration._
 class ActivationFileStorage(logFilePrefix: String,
                             logPath: Path,
                             writeResultToFile: Boolean,
-                            actorMaterializer: ActorMaterializer,
+                            actorSystem: ActorSystem,
                             logging: Logging) {
-
-  implicit val materializer = actorMaterializer
+  implicit val system: ActorSystem = actorSystem
 
   private var logFile = logPath
   private val bufferSize = 100.MB
   private val perms = EnumSet.of(OWNER_READ, OWNER_WRITE, GROUP_READ, GROUP_WRITE, OTHERS_READ, OTHERS_WRITE)
   private val writeToFile: Sink[ByteString, _] = MergeHub
     .source[ByteString]
     .batchWeighted(bufferSize.toBytes, _.length, identity)(_ ++ _)
-    .to(RestartSink.withBackoff(minBackoff = 1.seconds, maxBackoff = 60.seconds, randomFactor = 0.2) { () =>
-      LogRotatorSink(() => {
-        val maxSize = bufferSize.toBytes
-        var bytesRead = maxSize
-        element =>
-          {
-            val size = element.size
-
-            if (bytesRead + size > maxSize) {
-              logFile = logPath.resolve(s"$logFilePrefix-${Instant.now.toEpochMilli}.log")
-
-              logging.info(this, s"Rotating log file to '$logFile'")
-              createLogFile(logFile)
-              bytesRead = size
-              Some(logFile)
-            } else {
-              bytesRead += size
-              None
+    .to(RestartSink.withBackoff(RestartSettings(minBackoff = 1.seconds, maxBackoff = 60.seconds, randomFactor = 0.2)) {
+      () =>
+        LogRotatorSink(() => {
+          val maxSize = bufferSize.toBytes
+          var bytesRead = maxSize
+          element =>
+            {
+              val size = element.size
+
+              if (bytesRead + size > maxSize) {
+                logFile = logPath.resolve(s"$logFilePrefix-${Instant.now.toEpochMilli}.log")
+
+                logging.info(this, s"Rotating log file to '$logFile'")
+                createLogFile(logFile)
+                bytesRead = size
+                Some(logFile)
+              } else {
+                bytesRead += size
+                None
+              }
             }
-          }
-      })
+        })
     })
     .run()
 

common/scala/src/main/scala/org/apache/openwhisk/core/database/ActivationStore.scala

@@ -20,7 +20,6 @@ package org.apache.openwhisk.core.database
 import java.time.Instant
 
 import akka.actor.ActorSystem
-import akka.stream.ActorMaterializer
 import akka.http.scaladsl.model.HttpRequest
 import spray.json.JsObject
 import org.apache.openwhisk.common.{Logging, TransactionId}
@@ -199,5 +198,5 @@ trait ActivationStore {
 }
 
 trait ActivationStoreProvider extends Spi {
-  def instance(actorSystem: ActorSystem, actorMaterializer: ActorMaterializer, logging: Logging): ActivationStore
+  def instance(actorSystem: ActorSystem, logging: Logging): ActivationStore
 }