PMM-337 coordinator support stub

Author: trolley813

build.sbt

@@ -152,7 +152,7 @@ lazy val core = project.in(file("core"))
   .settings(commonSettings)
   .settings(
     libraryDependencies ++= Library.scalaTest ++ Library.logging ++ Library.config ++ Library.cats
-      ++ Library.jol.map(_ % "test") ++ Library.arrowDeps
+      ++ Library.jol.map(_ % "test")
   )
 
 lazy val config = project.in(file("config"))
@@ -176,7 +176,7 @@ lazy val http = project.in(file("http"))
   .settings(commonSettings)
   .settings(
     libraryDependencies ++= Library.scalaTest ++ Library.akka ++
-      Library.akkaHttp ++ Library.logging ++ Library.swayDB
+      Library.akkaHttp ++ Library.logging
   )
   .dependsOn(core, config, streaming, dsl)
 
@@ -195,12 +195,12 @@ lazy val itValid = project.in(file("integration/correctness"))
   )
   .dependsOn(core, streaming, http, config)
 
-lazy val itPerf = project.in(file("integration/performance"))
-  .settings(commonSettings)
-  .settings(
-    libraryDependencies ++= Library.scalaTest ++ Library.dbDrivers ++ Library.testContainers ++ Library.logging
-  )
-  .dependsOn(itValid)
+//lazy val itPerf = project.in(file("integration/performance"))
+//  .settings(commonSettings)
+//  .settings(
+//    libraryDependencies ++= Library.scalaTest ++ Library.dbDrivers ++ Library.testContainers ++ Library.logging
+//  )
+//  .dependsOn(itValid)
 
 
 /*** Other settings ***/

http/src/main/resources/application.conf

@@ -10,26 +10,10 @@ http {
   port = 8080
 }
 
-flink {
-  max-parallelism = 3000
-
-  metrics {
-    source = 0.numRecordsOut
-    search = 0.Flat_Map.currentEventTs
-    sink = 0.numRecordsOut
-  }
-
-  monitoring {
-    host = localhost
-    port = 8081
-  }
-
-  job-manager {
-    host = localhost
-    port = 8081
-  }
-
-  checkpointing-interval = 10000
+coordinator {
+  enabled = false
+  host = localhost
+  port = 8081
 }
 
 

http/src/main/scala/ru/itclover/tsp/http/HttpService.scala

@@ -28,25 +28,19 @@ trait HttpService extends RoutesProtocols {
 
   val blockingExecutorContext: ExecutionContextExecutor
 
-  val reporting: Option[JobReporting]
-
   private val configs = ConfigFactory.load()
   val isDebug = true
   val isHideExceptions = configs.getBoolean("general.is-hide-exceptions")
-  val flinkMonitoringHost: String = getEnvVarOrConfig("FLINK_MONITORING_HOST", "flink.monitoring.host")
-  val flinkMonitoringPort: Int = getEnvVarOrConfig("FLINK_MONITORING_PORT", "flink.monitoring.port").toInt
-  val monitoringUri: Uri = s"http://$flinkMonitoringHost:$flinkMonitoringPort"
 
   private val log = Logger[HttpService]
 
   def composeRoutes: Reader[ExecutionContextExecutor, Route] = {
     log.debug("composeRoutes started")
 
     val res = for {
-      jobs       <- JobsRoutes.fromExecutionContext(monitoringUri, reporting, blockingExecutorContext)
-      monitoring <- MonitoringRoutes.fromExecutionContext(monitoringUri)
+      jobs       <- JobsRoutes.fromExecutionContext(blockingExecutorContext)
       validation <- ValidationRoutes.fromExecutionContext()
-    } yield jobs ~ monitoring ~ validation
+    } yield jobs ~ validation
 
     log.debug("composeRoutes finished")
     res

http/src/main/scala/ru/itclover/tsp/http/Launcher.scala

@@ -4,11 +4,14 @@ import java.net.URLDecoder
 import java.util.concurrent.{SynchronousQueue, ThreadPoolExecutor, TimeUnit}
 import akka.actor.ActorSystem
 import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
 import akka.stream.ActorMaterializer
 import cats.implicits._
 import ru.itclover.tsp.RowWithIdx
-import ru.itclover.tsp.http.routes.JobReporting
 import ru.itclover.tsp.http.services.queuing.QueueManagerService
+
+import scala.concurrent.Future
+import scala.util.{Failure, Success}
 //import com.google.common.util.concurrent.ThreadFactoryBuilder
 import com.typesafe.config.ConfigFactory
 import com.typesafe.scalalogging.Logger
@@ -73,6 +76,25 @@ object Launcher extends App with HttpService {
   val bindingFuture = Http().bindAndHandle(route, host, port)
 
   log.info(s"Service online at http://$host:$port/" + (if (isDebug) " in debug mode." else ""))
+  val coordinator = getCoordinatorHostPort
+  coordinator.map {
+    case (enabled, host, port) => if (enabled) {
+      val uri = s"http://$host:$port/register"
+      log.warn(s"TSP coordinator connection enabled: connecting to $uri...")
+
+      val responseFuture: Future[HttpResponse] = Http().singleRequest(HttpRequest(uri = uri))
+
+      responseFuture
+        .onComplete {
+          case Success(res) => {
+            if (res.status.isFailure) sys.error(s"Error: TSP coordinator returned ${res.status}")
+          }
+          case Failure(ex)   => sys.error(s"Cannot connect to $uri: $ex")
+        }
+    } else {
+      log.warn("TSP coordinator connection disabled.")
+    }
+  }
 
   if (configs.getBoolean("general.is-follow-input")) {
     log.info("Press RETURN to stop...")
@@ -90,36 +112,23 @@ object Launcher extends App with HttpService {
     }
   }
 
-  def getClusterHostPort: Either[String, (String, Int)] = {
-    val host = getEnvVarOrConfig("FLINK_JOBMGR_HOST", "flink.job-manager.host")
-    val portStr = getEnvVarOrConfig("FLINK_JOBMGR_PORT", "flink.job-manager.port")
+  def getCoordinatorHostPort: Either[String, (Boolean, String, Int)] = {
+    val enabledStr = getEnvVarOrConfig("COORDINATOR_ENABLED", "coordinator.enabled")
+    val host = getEnvVarOrConfig("COORDINATOR_HOST", "coordinator.host")
+    val portStr = getEnvVarOrConfig("COORDINATOR_PORT", "coordinator.port")
     val port = Either.catchNonFatal(portStr.toInt).left.map { ex: Throwable =>
-      s"Cannot parse FLINK_JOBMGR_PORT ($portStr): ${ex.getMessage}"
+      s"Cannot parse COORDINATOR_PORT ($portStr): ${ex.getMessage}"
     }
-    port.map(p => (host, p))
-  }
-
-  override val reporting: Option[JobReporting] = {
-    val reportingEnabledConfig = getEnvVarOrNone("JOB_REPORTING_ENABLED").getOrElse("0")
-    val reportingEnabled = reportingEnabledConfig match {
-      case "0" | "false" | "off" | "no" => false
-      case "1" | "true" | "on" | "yes"  => true
-      case _                            => sys.error(s"JOB_REPORTING_ENABLED not set or set to a unsupported value: $reportingEnabledConfig")
-    }
-
-    if (reportingEnabled) {
-      val reportingBroker = getEnvVarOrNone("JOB_REPORTING_BROKER")
-        .getOrElse(sys.error("Job reporting enabled, but JOB_REPORTING_BROKER not set"))
-      val reportingTopic = getEnvVarOrNone("JOB_REPORTING_TOPIC")
-        .getOrElse(sys.error("Job reporting enabled, but JOB_REPORTING_TOPIC not set"))
-      log.warn(s"Job reporting enabled, sending to topic $reportingTopic on $reportingBroker")
-      Some(JobReporting(reportingBroker, reportingTopic))
-    } else {
-      log.warn("Job reporting disabled")
-      None
+    val enabled = Either.catchNonFatal(enabledStr.toBoolean) match {
+      case Left(ex) => {
+        log.warn(s"Cannot parse COORDINATOR_ENABLED ($enabledStr), defaulting to false:  ${ex.getMessage}")
+        false
+      }
+      case Right(value) => value
     }
+    port.map(p => (enabled, host, p))
   }
 
-  implicit val rep = reporting
-  val queueManager = QueueManagerService.getOrCreate(monitoringUri, blockingExecutorContext)
+
+  val queueManager = QueueManagerService.getOrCreate("mgr", blockingExecutorContext)
 }

http/src/main/scala/ru/itclover/tsp/http/routes/JobsRoutes.scala

@@ -23,29 +23,22 @@ import ru.itclover.tsp.http.domain.output.SuccessfulResponse.ExecInfo
 import ru.itclover.tsp.http.domain.output._
 import ru.itclover.tsp.http.protocols.RoutesProtocols
 import ru.itclover.tsp.http.services.queuing.QueueManagerService
-import ru.itclover.tsp.http.services.streaming.FlinkMonitoringService
 import ru.itclover.tsp.streaming.io.{InputConf, JDBCInputConf, KafkaInputConf}
 import ru.itclover.tsp.streaming.io.{JDBCOutputConf, KafkaOutputConf, OutputConf}
 import ru.itclover.tsp.streaming.mappers._
 
 import scala.concurrent.{ExecutionContextExecutor, Future}
 import ru.itclover.tsp.streaming.utils.ErrorsADT.{ConfigErr, Err, GenericRuntimeErr, RuntimeErr}
 
-case class JobReporting(brokers: String, topic: String)
-
 trait JobsRoutes extends RoutesProtocols {
   implicit val executionContext: ExecutionContextExecutor
   val blockingExecutionContext: ExecutionContextExecutor
   implicit val actorSystem: ActorSystem
   implicit val materializer: ActorMaterializer
   implicit val decoders = AnyDecodersInstances
 
-  val monitoringUri: Uri
-  lazy val monitoring = FlinkMonitoringService(monitoringUri)
   val queueManager: QueueManagerService
 
-  implicit val reporting: Option[JobReporting]
-
   private val log = Logger[JobsRoutes]
 
   val route: Route =
@@ -72,9 +65,7 @@ object JobsRoutes {
 
   private val log = Logger[JobsRoutes]
 
-  def fromExecutionContext(monitoringUrl: Uri,
-                           rep: Option[JobReporting],
-                           blocking: ExecutionContextExecutor)(
+  def fromExecutionContext(blocking: ExecutionContextExecutor)(
     implicit as: ActorSystem,
     am: ActorMaterializer
   ): Reader[ExecutionContextExecutor, Route] = {
@@ -87,11 +78,9 @@ object JobsRoutes {
         implicit val executionContext: ExecutionContextExecutor = execContext
         implicit val actorSystem = as
         implicit val materializer = am
-        override val monitoringUri = monitoringUrl
-        override val queueManager = QueueManagerService.getOrCreate(monitoringUrl, blocking)(
-          execContext, as, am, AnyDecodersInstances, rep
+        override val queueManager = QueueManagerService.getOrCreate("mgr", blocking)(
+          execContext, as, am, AnyDecodersInstances
         )
-        override val reporting: Option[JobReporting] = rep
       }.route
     }