From 8600fc55684d805596523cf10270011d41cf82e1 Mon Sep 17 00:00:00 2001 From: Octav Zaharia Date: Sat, 25 Jun 2022 09:58:55 +0300 Subject: [PATCH] Upgrade to ZIO 2 --- build.sbt | 12 +- .../io/km8/core/kafka/KafkaExplorerSpec.scala | 2 - .../io/km8/core/kafka/KafkaProducerSpec.scala | 15 +- .../it/scala/io/km8/core/kafka/itlayers.scala | 9 +- .../io/km8/core/config/ClustersConfig.scala | 21 +- .../km8/core/config/ConfigPathService.scala | 9 +- .../io/km8/core/kafka/KafkaConsumer.scala | 40 +-- .../io/km8/core/kafka/KafkaExplorer.scala | 240 +++++++++--------- .../io/km8/core/kafka/KafkaProducer.scala | 81 +++--- .../main/scala/io/km8/core/utils/Logger.scala | 8 +- .../test/scala/io/kafkamate/BasicSpec.scala | 7 +- .../kafka/consumer/ConsumerSpec.scala | 4 +- .../scala/io/kafkamate/util/HelperSpec.scala | 10 +- .../io/kafkamate/util/KafkaEmbedded.scala | 2 +- fx/src/main/scala/io/km8/fx/Main.scala | 23 +- fx/src/main/scala/io/km8/fx/models/UI.scala | 44 ++-- .../main/scala/io/km8/fx/ui/BaseControl.scala | 2 +- .../fx/ui/components/ClusterListControl.scala | 4 +- .../km8/fx/ui/components/HeaderControl.scala | 2 +- .../fx/ui/components/MainContentControl.scala | 6 +- .../fx/ui/components/NavigatorControl.scala | 8 +- .../main/scala/io/km8/fx/views/Clusters.scala | 3 + project/plugins.sbt | 2 + 23 files changed, 278 insertions(+), 276 deletions(-) create mode 100644 fx/src/main/scala/io/km8/fx/views/Clusters.scala diff --git a/build.sbt b/build.sbt index 3191a48..5355422 100644 --- a/build.sbt +++ b/build.sbt @@ -3,15 +3,15 @@ ThisBuild / resolvers += Resolver.sonatypeRepo("snapshots") lazy val ProjectName = "KM8" lazy val ProjectOrganization = "KM8S" lazy val ProjectVersion = "0.2.0-SNAPSHOT" -lazy val ProjectScalaVersion = "3.1.2" +lazy val ProjectScalaVersion = "3.1.3" lazy val Versions = new { - val zio = "1.0.13" - val zioKafka = "0.17.5" - val zioJson = "0.2.0-M3" - val zioLogging = "0.5.14" - val zioPrelude = "1.0.0-RC8" + val zio = "2.0.0-RC5" + val zioKafka = "2.0.0-M3" + val zioJson = "0.3.0-RC7" + val zioLogging = "2.0.0-RC8" + val zioPrelude = "1.0.0-RC13" val kafkaProtobuf = "7.1.0" val javaFx = "16" diff --git a/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala b/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala index 569b303..a8c32ee 100644 --- a/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala +++ b/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala @@ -1,7 +1,6 @@ package io.km8.core.kafka import zio.* -import zio.blocking.Blocking import zio.clock.Clock import zio.duration.* import zio.kafka.consumer.* @@ -22,7 +21,6 @@ object KafkaExplorerSpec extends DefaultRunnableSpec: val layer = Clock.live >+> - Blocking.live >+> itlayers.kafkaContainer >+> itlayers.clusterConfig(clusterId) >+> KafkaExplorer.liveLayer >+> diff --git a/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala b/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala index 818c41a..400a185 100644 --- a/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala +++ b/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala @@ -1,7 +1,6 @@ package io.km8.core.kafka import zio.* -import zio.blocking.Blocking import zio.clock.Clock import zio.duration.* import zio.kafka.consumer.* @@ -16,19 +15,19 @@ import org.apache.kafka.clients.consumer.OffsetResetStrategy object KafkaProducerSpec extends DefaultRunnableSpec: - private val consumerLayer: ZLayer[Has[KafkaContainer], Nothing, Has[KafkaConsumer]] = - Clock.live ++ Blocking.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaConsumer.liveLayer + private val consumerLayer: ZLayer[KafkaContainer, Nothing, KafkaConsumer] = + Clock.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaConsumer.liveLayer - private val producerLayer: ZLayer[Has[KafkaContainer], Nothing, KafkaProducer] = - Blocking.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaProducer.liveLayer + private val producerLayer: ZLayer[KafkaContainer, Nothing, KafkaProducer] = + itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaProducer.liveLayer - val specLayer: ZLayer[Has[KafkaContainer], Nothing, Has[KafkaConsumer] & KafkaProducer] = + val specLayer: ZLayer[KafkaContainer, Nothing, KafkaConsumer & KafkaProducer] = consumerLayer ++ producerLayer override def spec: ZSpec[_root_.zio.test.environment.TestEnvironment, Any] = mainSpec - .provideSomeLayer[environment.TestEnvironment & Has[KafkaContainer]](specLayer ++ Clock.live) - .provideCustomLayerShared(Blocking.live >>> itlayers.kafkaContainer) + .provideSomeLayer[environment.TestEnvironment & KafkaContainer](specLayer ++ Clock.live) + .provideCustomLayerShared(itlayers.kafkaContainer) private val mainSpec = suite("Kafka services")( diff --git a/core/src/it/scala/io/km8/core/kafka/itlayers.scala b/core/src/it/scala/io/km8/core/kafka/itlayers.scala index ff96bfe..ecbc6ed 100644 --- a/core/src/it/scala/io/km8/core/kafka/itlayers.scala +++ b/core/src/it/scala/io/km8/core/kafka/itlayers.scala @@ -3,7 +3,6 @@ package io.km8.core.kafka import com.dimafeng.testcontainers.KafkaContainer import zio.* import zio.clock.Clock -import zio.blocking.* import zio.kafka.consumer.Consumer import zio.kafka.consumer.Consumer.{AutoOffsetStrategy, OffsetRetrieval} import zio.kafka.consumer.ConsumerSettings @@ -12,7 +11,7 @@ import io.km8.core.config.{ClusterConfig, ClusterProperties, ClusterSettings} object itlayers: - val kafkaContainer: ZLayer[Blocking, Nothing, Has[KafkaContainer]] = + val kafkaContainer: ZLayer[Any, Nothing, KafkaContainer] = ZManaged.make { effectBlocking { val container = new KafkaContainer() @@ -21,7 +20,7 @@ object itlayers: }.orDie }(container => effectBlocking(container.stop()).orDie).toLayer - def consumerSettings(cg: String): ZManaged[Has[KafkaContainer], Nothing, ConsumerSettings] = + def consumerSettings(cg: String): ZManaged[KafkaContainer, Nothing, ConsumerSettings] = ZIO .service[KafkaContainer] .map(c => @@ -31,10 +30,10 @@ object itlayers: ) .toManaged_ - def consumerLayer(cgroup: String): ZLayer[Has[KafkaContainer] with Clock with Blocking, Nothing, Has[Consumer]] = + def consumerLayer(cgroup: String): ZLayer[KafkaContainer with Clock , Nothing, Consumer] = consumerSettings(cgroup).flatMap(Consumer.make(_)).orDie.toLayer - def clusterConfig(clusterId: String): ZLayer[Has[KafkaContainer], Nothing, Has[ClusterConfig]] = + def clusterConfig(clusterId: String): ZLayer[KafkaContainer, Nothing, ClusterConfig] = ZIO .service[KafkaContainer] .map(kafkaContainer => diff --git a/core/src/main/scala/io/km8/core/config/ClustersConfig.scala b/core/src/main/scala/io/km8/core/config/ClustersConfig.scala index 3f5a18f..999d789 100644 --- a/core/src/main/scala/io/km8/core/config/ClustersConfig.scala +++ b/core/src/main/scala/io/km8/core/config/ClustersConfig.scala @@ -48,29 +48,30 @@ trait ClusterConfig { object ClustersConfig { - def readClusters = ZIO.serviceWith[ClusterConfig](_.readClusters) - def writeClusters(cluster: ClusterSettings) = ZIO.serviceWith[ClusterConfig](_.writeClusters(cluster)) - def deleteCluster(clusterId: String) = ZIO.serviceWith[ClusterConfig](_.deleteCluster(clusterId)) + def readClusters = ZIO.serviceWithZIO[ClusterConfig](_.readClusters) + def writeClusters(cluster: ClusterSettings) = ZIO.serviceWithZIO[ClusterConfig](_.writeClusters(cluster)) + def deleteCluster(clusterId: String) = ZIO.serviceWithZIO[ClusterConfig](_.deleteCluster(clusterId)) - lazy val liveLayer: URLayer[Has[ConfigPath] with Logging, Has[ClusterConfig]] = (ClusterConfigLive(_, _)).toLayer + lazy val liveLayer: URLayer[ConfigPath , ClusterConfig] = + ZLayer(ZIO.service[ConfigPath].map(ClusterConfigLive.apply)) - case class ClusterConfigLive(configPath: ConfigPath, log: Logger[String]) extends ClusterConfig { + case class ClusterConfigLive(configPath: ConfigPath) extends ClusterConfig { private val configFilepath = configPath.path private def emptyProperties = ClusterProperties(List.empty) private def emptyPropertiesJson = emptyProperties.toJsonPretty private def writeJson(json: => String) = - Task(os.write.over(configFilepath, json, createFolders = true)) + ZIO.attempt(os.write.over(configFilepath, json, createFolders = true)) def readClusters: Task[ClusterProperties] = for { - b <- Task(os.exists(configFilepath)) + b <- ZIO.attempt(os.exists(configFilepath)) _ <- writeJson(emptyPropertiesJson).unless(b) - s <- Task(os.read(configFilepath)) + s <- ZIO.attempt(os.read(configFilepath)) r <- ZIO .fromEither(s.fromJson[ClusterProperties]) .catchAll { err => - log.warn(s"Parsing error: $err") *> + ZIO.logWarning(s"Parsing error: $err") *> writeJson(emptyPropertiesJson).as(emptyProperties) } } yield r @@ -85,7 +86,7 @@ object ClustersConfig { def deleteCluster(clusterId: String): Task[ClusterProperties] = for { c <- readClusters - ls <- ZIO.filterNotPar(c.clusters)(s => Task(s.id == clusterId)) + ls <- ZIO.filterNotPar(c.clusters)(s => ZIO.attempt(s.id == clusterId)) json = ClusterProperties(ls).toJsonPretty _ <- writeJson(json) r <- readClusters diff --git a/core/src/main/scala/io/km8/core/config/ConfigPathService.scala b/core/src/main/scala/io/km8/core/config/ConfigPathService.scala index 6749268..4fa1f73 100644 --- a/core/src/main/scala/io/km8/core/config/ConfigPathService.scala +++ b/core/src/main/scala/io/km8/core/config/ConfigPathService.scala @@ -2,15 +2,16 @@ package io.km8.core package config import zio.* -import zio.system.* +import zio.System +import zio.System.env lazy val EnvKey = "KAFKAMATE_ENV" lazy val FileName = "kafkamate.json" case class ConfigPath(path: os.Path) -lazy val liveLayer: URLayer[System, Has[ConfigPath]] = - env(EnvKey).map { +lazy val liveLayer: URLayer[System, ConfigPath] = + ZLayer.fromZIO(env(EnvKey).map { case Some("prod") => ConfigPath(os.root / FileName) case _ => ConfigPath(os.pwd / FileName) - }.orDie.toLayer + }.orDie) diff --git a/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala b/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala index f1dddc4..19721b8 100644 --- a/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala +++ b/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala @@ -2,9 +2,9 @@ package io.km8.core package kafka import zio.* -import zio.blocking.Blocking -import zio.clock.Clock -import zio.duration.* + +import zio.Clock + import zio.kafka.consumer.* import zio.kafka.consumer.Consumer.* import zio.kafka.serde.Deserializer @@ -28,19 +28,23 @@ trait KafkaConsumer { object KafkaConsumer { - lazy val liveLayer: URLayer[Clock with Blocking with Has[ClusterConfig], Has[KafkaConsumer]] = - (KafkaConsumerLive(_, _, _)).toLayer + lazy val liveLayer: URLayer[Clock with ClusterConfig, KafkaConsumer] = + ZLayer { + for { + clock <- ZIO.service[Clock] + config <- ZIO.service[ClusterConfig] + } yield KafkaConsumerLive(clock, config) + } - def consume(request: ConsumeRequest): ZStream[Has[KafkaConsumer], Throwable, Message] = - ZStream.accessStream[Has[KafkaConsumer]](_.get.consume(request)) + def consume(request: ConsumeRequest): ZStream[KafkaConsumer, Throwable, Message] = + ZStream.environmentWithStream[KafkaConsumer](_.get.consume(request)) case class KafkaConsumerLive( - clock: Clock.Service, - blocking: Blocking.Service, + clock: Clock, clustersConfigService: ClusterConfig) extends KafkaConsumer { + private val clockLayer = ZLayer.succeed(clock) - private val blockingLayer = ZLayer.succeed(blocking) // TODO Ciprian transform to enum or reuse the Consumer enum + serdes private def extractOffsetStrategy(offsetValue: String): AutoOffsetStrategy = @@ -55,7 +59,7 @@ object KafkaConsumer { .map(_.asTry) private def consumerSettings(config: ClusterSettings, offsetStrategy: String): Task[ConsumerSettings] = - Task { + ZIO.attempt { val uuid = UUID.randomUUID().toString ConsumerSettings(config.kafkaHosts) .withGroupId(s"group-kafkamate-$uuid") @@ -68,11 +72,11 @@ object KafkaConsumer { private def makeConsumerLayer( clusterId: String, offsetStrategy: String - ): ZLayer[Clock & Blocking, Throwable, Has[Consumer]] = - ZLayer.fromManaged { + ): ZLayer[Any, Throwable, Consumer] = + ZLayer.scoped { for { - cs <- clustersConfigService.getCluster(clusterId).toManaged_ - settings <- consumerSettings(cs, offsetStrategy).toManaged_ + cs <- clustersConfigService.getCluster(clusterId) + settings <- consumerSettings(cs, offsetStrategy) consumer <- Consumer.make(settings) } yield consumer } @@ -80,7 +84,7 @@ object KafkaConsumer { def consume(request: ConsumeRequest): ZStream[Any, Throwable, Message] = { def consumer[T]( valueDeserializer: Deserializer[Any, Try[T]] - ): ZStream[Has[Consumer], Throwable, Message] = Consumer + ): ZStream[Consumer, Throwable, Message] = Consumer .subscribeAnd(Subscription.topics(request.topicName)) .plainStream(Deserializer.string, valueDeserializer) .collect { @@ -98,7 +102,7 @@ object KafkaConsumer { .orElseFail(new Exception("SchemaRegistry url was not provided!")) ) ZStream - .fromEffect(protoSettings.flatMap(protobufDeserializer)) + .fromZIO(protoSettings.flatMap(protobufDeserializer)) .flatMap(consumer) case _ => consumer(Deserializer.string.asTry) } @@ -114,7 +118,7 @@ object KafkaConsumer { else withFilter.take(request.maxResults) withFilterLimit.provideLayer( - clockLayer ++ blockingLayer >>> + clockLayer >>> makeConsumerLayer(request.clusterId, request.offsetStrategy) ) } diff --git a/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala b/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala index 2fe4b2a..3316f66 100644 --- a/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala +++ b/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala @@ -2,9 +2,9 @@ package io.km8.core package kafka import zio.* -import zio.blocking.Blocking -import zio.clock.Clock -import zio.duration.* + +import zio.Clock + import zio.kafka.admin.* import zio.kafka.admin.AdminClient.* @@ -24,126 +24,130 @@ object KafkaExplorer extends Accessible[KafkaExplorer]: val CleanupPolicyKey = "cleanup.policy" val RetentionMsKey = "retention.ms" - lazy val liveLayer: URLayer[Has[ClusterConfig] with Clock with Blocking, Has[KafkaExplorer]] = - KafkaExplorerLive.apply.toLayer - - case class KafkaExplorerLive( - clock: Clock.Service, - blocking: Blocking.Service, - clustersConfigService: ClusterConfig) - extends KafkaExplorer { - val clockLayer = ZLayer.succeed(clock) - val blockingLayer = ZLayer.succeed(blocking) - - private def adminClientLayer(clusterId: String) = - ZLayer.fromManaged { - for { - cs <- clustersConfigService.getCluster(clusterId).toManaged_ - client <- AdminClient.make(AdminClientSettings(cs.kafkaHosts, 2.seconds, Map.empty)) - } yield client - } - - def withAdminClient[A](clusterId: String)(eff: AdminClient => RIO[Blocking with Clock, A]): Task[A] = + lazy val liveLayer: ZLayer[Clock & ClusterConfig, Nothing, KafkaExplorer] = + ZLayer{ + for { + cl <- ZIO.service[Clock] + cc <- ZIO.service[ClusterConfig] + } yield KafkaExplorerLive(cl, cc) + } + + +case class KafkaExplorerLive(clock: Clock, clustersConfigService: ClusterConfig) extends KafkaExplorer { + val clockLayer: ULayer[Clock] = ZLayer.succeed(clock) + + private def adminClientLayer(clusterId: String) = + ZLayer.fromZIO { + for { + cs <- clustersConfigService.getCluster(clusterId) + client <- AdminClient.make(AdminClientSettings(cs.kafkaHosts, 2.seconds, Map.empty)) + } yield client + } + + def withAdminClient[A](clusterId: String)(eff: AdminClient => RIO[Clock, A]): ZIO[Any, Throwable, A] = + ZIO.scoped { ZIO .service[AdminClient] .flatMap(eff(_).timeoutFail(new Exception("Timed out"))(6.seconds)) - .provideLayer(blockingLayer >+> adminClientLayer(clusterId) ++ clockLayer) - - override def listBrokers(clusterId: String) = - withAdminClient(clusterId) { ac => - for { - (nodes, controllerId) <- ac.describeClusterNodes() <&> ac.describeClusterController().map(_.map(_.id)) - brokers = nodes.map { n => - val nodeId = n.id - if (controllerId.contains(nodeId)) BrokerDetails(nodeId, isController = true) - else BrokerDetails(id = nodeId, isController = false) - } - // resources = nodes.map(n => new ConfigResource(ConfigResource.Type.BROKER, n.idString())) - // _ <- ac.describeConfigs(resources) - } yield brokers - } - - override def listTopics(clusterId: String) = - withAdminClient(clusterId) { ac => - ac.listTopics() - .map(_.keys.toList) - .flatMap(ls => ZIO.filterNotPar(ls)(t => UIO(t.startsWith("_")))) - .flatMap(ls => - ac.describeTopics(ls) <&> ac.describeConfigs(ls.map(ConfigResource(ConfigResourceType.Topic, _))) - ) - .map { case (nameDescriptionMap, topicConfigMap) => - val configs = topicConfigMap.map { case (res, conf) => (res.name, conf) } - nameDescriptionMap.map { case (name, description) => - val conf = configs.get(name).map(_.entries) - def getConfig(key: String) = conf.flatMap(_.get(key).map(_.value())).getOrElse("unknown") - TopicDetails( - name = name, - partitions = description.partitions.size, - replication = description.partitions.headOption.map(_.replicas.size).getOrElse(0), - cleanupPolicy = getConfig(CleanupPolicyKey), - retentionMs = getConfig(RetentionMsKey), - size = 0 - ) - }.toList.sortBy(_.name) - } - } - - override def addTopic(req: AddTopicRequest) = - withAdminClient(req.clusterId) { ac => - ac - .createTopic( - AdminClient.NewTopic( - req.name, - req.partitions, - req.replication.toShort, - Map( - CleanupPolicyKey -> req.cleanupPolicy, - RetentionMsKey -> req.retentionMs - ) - ) - ) - .as( + .provideLayer(adminClientLayer(clusterId) ++ clockLayer) + } + + override def listBrokers(clusterId: String) = + withAdminClient(clusterId) { ac => + for { + nodes <- ac.describeClusterNodes() + controllerId <- ac.describeClusterController().map(_.map(_.id)) + brokers = nodes.map { n => + val nodeId = n.id + if (controllerId.contains(nodeId)) BrokerDetails(nodeId, isController = true) + else BrokerDetails(id = nodeId, isController = false) + } + // resources = nodes.map(n => new ConfigResource(ConfigResource.Type.BROKER, n.idString())) + // _ <- ac.describeConfigs(resources) + } yield brokers + } + + override def listTopics(clusterId: String) = + withAdminClient(clusterId) { ac => + ac.listTopics() + .map(_.keys.toList) + .flatMap(ls => ZIO.filterNotPar(ls)(t => ZIO.succeed(t.startsWith("_")))) + .flatMap(ls => + ac.describeTopics(ls) <&> ac.describeConfigs(ls.map(ConfigResource(ConfigResourceType.Topic, _))) + ) + .map { case (nameDescriptionMap, topicConfigMap) => + val configs = topicConfigMap.map { case (res, conf) => (res.name, conf) } + nameDescriptionMap.map { case (name, description) => + val conf = configs.get(name).map(_.entries) + def getConfig(key: String) = conf.flatMap(_.get(key).map(_.value())).getOrElse("unknown") TopicDetails( - name = req.name, - partitions = req.partitions, - replication = req.replication, - cleanupPolicy = req.cleanupPolicy, - retentionMs = req.retentionMs, + name = name, + partitions = description.partitions.size, + replication = description.partitions.headOption.map(_.replicas.size).getOrElse(0), + cleanupPolicy = getConfig(KafkaExplorer.CleanupPolicyKey), + retentionMs = getConfig(KafkaExplorer.RetentionMsKey), size = 0 ) + }.toList.sortBy(_.name) + } + } + + override def addTopic(req: AddTopicRequest) = + withAdminClient(req.clusterId) { ac => + ac + .createTopic( + AdminClient.NewTopic( + req.name, + req.partitions, + req.replication.toShort, + Map( + KafkaExplorer.CleanupPolicyKey -> req.cleanupPolicy, + KafkaExplorer.RetentionMsKey -> req.retentionMs + ) ) - } - - override def deleteTopic(req: DeleteTopicRequest) = - withAdminClient(req.clusterId) { - _.deleteTopic(req.topicName) - .as(DeleteTopicResponse(req.topicName)) - } - - override def listConsumerGroups(clusterId: String): Task[ConsumerGroupsResponse] = - def mapConsumerGroup(state: Option[ConsumerGroupState]): ConsumerGroupInternalState = state match { - case None | Some(ConsumerGroupState.Unknown) => ConsumerGroupInternalState.Unknown - case Some(ConsumerGroupState.PreparingRebalance) => ConsumerGroupInternalState.PreparingRebalance - case Some(ConsumerGroupState.CompletingRebalance) => ConsumerGroupInternalState.CompletingRebalance - case Some(ConsumerGroupState.Stable) => ConsumerGroupInternalState.Stable - case Some(ConsumerGroupState.Dead) => ConsumerGroupInternalState.Dead - case Some(ConsumerGroupState.Empty) => ConsumerGroupInternalState.Empty - } - - withAdminClient(clusterId) { - _.listConsumerGroups() - .map(lst => ConsumerGroupsResponse(lst.map(r => ConsumerGroupInternal(r.groupId, mapConsumerGroup(r.state))))) - } - - end listConsumerGroups - - override def listConsumerOffsets(clusterId: String, groupId: String): Task[ConsumerGroupOffsetsResponse] = - withAdminClient(clusterId) { - _.listConsumerGroupOffsets(groupId).map(res => - ConsumerGroupOffsetsResponse(res.map { case (tp, offMeta) => - TopicPartitionInternal(tp.name, tp.partition) -> offMeta.offset - }) ) - } - - } + .as( + TopicDetails( + name = req.name, + partitions = req.partitions, + replication = req.replication, + cleanupPolicy = req.cleanupPolicy, + retentionMs = req.retentionMs, + size = 0 + ) + ) + } + + override def deleteTopic(req: DeleteTopicRequest) = + withAdminClient(req.clusterId) { + _.deleteTopic(req.topicName) + .as(DeleteTopicResponse(req.topicName)) + } + + override def listConsumerGroups(clusterId: String): Task[ConsumerGroupsResponse] = + def mapConsumerGroup(state: Option[ConsumerGroupState]): ConsumerGroupInternalState = state match { + case None | Some(ConsumerGroupState.Unknown) => ConsumerGroupInternalState.Unknown + case Some(ConsumerGroupState.PreparingRebalance) => ConsumerGroupInternalState.PreparingRebalance + case Some(ConsumerGroupState.CompletingRebalance) => ConsumerGroupInternalState.CompletingRebalance + case Some(ConsumerGroupState.Stable) => ConsumerGroupInternalState.Stable + case Some(ConsumerGroupState.Dead) => ConsumerGroupInternalState.Dead + case Some(ConsumerGroupState.Empty) => ConsumerGroupInternalState.Empty + } + + withAdminClient(clusterId) { + _.listConsumerGroups() + .map(lst => ConsumerGroupsResponse(lst.map(r => ConsumerGroupInternal(r.groupId, mapConsumerGroup(r.state))))) + } + + end listConsumerGroups + + override def listConsumerOffsets(clusterId: String, groupId: String): ZIO[Any, Throwable,ConsumerGroupOffsetsResponse] = + withAdminClient(clusterId) { + _.listConsumerGroupOffsets(groupId).map(res => + ConsumerGroupOffsetsResponse(res.map { case (tp, offMeta) => + TopicPartitionInternal(tp.name, tp.partition) -> offMeta.offset + }) + ) + } + +} diff --git a/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala b/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala index 6c2e2b4..c82628b 100644 --- a/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala +++ b/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala @@ -2,26 +2,25 @@ package io.km8.core package kafka import zio.* -import zio.blocking.* + import zio.kafka.producer.* import zio.kafka.serde.* import io.km8.core.config.* import io.km8.core.config.ClustersConfig.* -object KafkaProducer { - type KafkaProducer = Has[Service] +trait KafkaProducer { - trait Service { + def produce( + topic: String, + key: String, + value: String + )( + clusterId: String + ): ZIO[Scope, Throwable, Unit] +} - def produce( - topic: String, - key: String, - value: String - )( - clusterId: String - ): Task[Unit] - } +object KafkaProducer { def produce( topic: String, @@ -29,39 +28,35 @@ object KafkaProducer { value: String )( clusterId: String - ): RIO[KafkaProducer, Unit] = ZIO.accessM[KafkaProducer](_.get.produce(topic, key, value)(clusterId)) + ): ZIO[Scope & KafkaProducer, Throwable, Unit] = ZIO.environmentWithZIO[KafkaProducer](_.get.produce(topic, key, value)(clusterId)) - lazy val liveLayer: URLayer[Has[ClusterConfig] with Blocking, KafkaProducer] = - ZLayer.fromServices[ClusterConfig, Blocking.Service, KafkaProducer.Service]( - (clusterConfigService: ClusterConfig, blocking: Blocking.Service) => - new Service { + lazy val liveLayer: URLayer[ClusterConfig & Scope, KafkaProducer] =ZLayer.fromZIO( for { + clusterConfigService <- ZIO.service[ClusterConfig] + } yield new KafkaProducer { - lazy val serdeLayer: ULayer[Has[Serializer[Any, String]]] = - ZLayer.succeed(Serde.string) + lazy val serdeLayer: ULayer[Serializer[Any, String]] = + ZLayer.succeed(Serde.string) - def settingsLayer(clusterId: String): Task[ProducerSettings] = - clusterConfigService - .getCluster(clusterId) - .map(c => ProducerSettings(c.kafkaHosts)) + def settingsLayer(clusterId: String): Task[ProducerSettings] = + clusterConfigService + .getCluster(clusterId) + .map(c => ProducerSettings(c.kafkaHosts)) - def producerLayer(clusterId: String): ZLayer[Any, Throwable, Has[Producer]] = - settingsLayer(clusterId).toManaged_ - .flatMap(settings => Producer.make(settings)) - .provideLayer(ZLayer.succeed(blocking) ++ serdeLayer ++ ZLayer.succeed(clusterConfigService)) - .toLayer + def producerLayer(clusterId: String): ZIO[Scope, Throwable, Producer] = + settingsLayer(clusterId) + .flatMap(settings => Producer.make(settings)) - def produce( - topic: String, - key: String, - value: String - )( - clusterId: String - ): Task[Unit] = - Producer - .produce[Any, String, String](topic, key, value, Serde.string, Serde.string) - .unit - .provideLayer(producerLayer(clusterId)) - } - ) - -} + def produce( + topic: String, + key: String, + value: String + )( + clusterId: String + ): ZIO[Scope, Throwable, Unit] = + Producer + .produce[Any, String, String](topic, key, value, Serde.string, Serde.string) + .unit + .provideLayer(ZLayer.fromZIO(producerLayer(clusterId))) + } + ) +} \ No newline at end of file diff --git a/core/src/main/scala/io/km8/core/utils/Logger.scala b/core/src/main/scala/io/km8/core/utils/Logger.scala index b064390..3936324 100644 --- a/core/src/main/scala/io/km8/core/utils/Logger.scala +++ b/core/src/main/scala/io/km8/core/utils/Logger.scala @@ -2,16 +2,18 @@ package io.km8.core package utils import zio.URLayer -import zio.clock.Clock -import zio.console.Console +import zio.Clock import zio.logging.* +import zio.Console object Logger { +/* lazy val liveLayer: URLayer[Console with Clock, Logging] = - Logging.console( + console( logLevel = LogLevel.Info, format = LogFormat.ColoredLogFormat() ) >>> Logging.withRootLoggerName("kafkamate") +*/ } diff --git a/core/src/test/scala/io/kafkamate/BasicSpec.scala b/core/src/test/scala/io/kafkamate/BasicSpec.scala index 9456aca..cb94812 100644 --- a/core/src/test/scala/io/kafkamate/BasicSpec.scala +++ b/core/src/test/scala/io/kafkamate/BasicSpec.scala @@ -3,15 +3,16 @@ package io.kafkamate import zio.ZIO import zio.test.* import zio.test.Assertion.equalTo +import zio.test.ZIOSpecDefault /** * This is a sample test for the CI; TODO delete after we start creating real tests for the app */ -object BasicSpec extends DefaultRunnableSpec { +object BasicSpec extends ZIOSpecDefault { - override def spec: ZSpec[_root_.zio.test.environment.TestEnvironment, Any] = + override def spec = suite("Simple test")( - testM("Check zio result") { + test("Check zio result") { assertM(ZIO.succeed(1))(equalTo(1)) } ) diff --git a/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala b/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala index 66fe5fe..f3aa435 100644 --- a/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala +++ b/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala @@ -4,7 +4,6 @@ package consumer /* import zio.* -import zio.blocking.Blocking import zio.clock.Clock import zio.duration.* import zio.test.Assertion.* @@ -22,10 +21,9 @@ object ConsumerSpec extends DefaultRunnableSpec with HelperSpec { import messages.* val testLayer - : Layer[TestFailure[Throwable], Clock with Blocking with Logging with StringProducer with KafkaConsumer] = + : Layer[TestFailure[Throwable], Clock with StringProducer with KafkaConsumer] = (Clock.live >+> Console.live >+> - Blocking.live >+> KafkaEmbedded.Kafka.embedded >+> stringProducer >+> testConfigLayer >+> diff --git a/core/src/test/scala/io/kafkamate/util/HelperSpec.scala b/core/src/test/scala/io/kafkamate/util/HelperSpec.scala index be630a4..4835283 100644 --- a/core/src/test/scala/io/kafkamate/util/HelperSpec.scala +++ b/core/src/test/scala/io/kafkamate/util/HelperSpec.scala @@ -4,7 +4,6 @@ package util /* import org.apache.kafka.clients.producer.{ProducerRecord, RecordMetadata} import zio.* -import zio.blocking.Blocking import zio.clock.Clock import zio.kafka.producer.{Producer, ProducerSettings} import zio.kafka.serde.{Serde, Serializer} @@ -18,13 +17,12 @@ trait HelperSpec { val producerSettings: URIO[Kafka, ProducerSettings] = ZIO.access[Kafka](_.get.bootstrapServers).map(ProducerSettings(_)) - val stringProducer: ZLayer[Kafka with Blocking, Throwable, StringProducer] = - (Blocking.any ++ producerSettings.toLayer ++ ZLayer.succeed(Serde.string: Serializer[Any, String])) >>> + val stringProducer: ZLayer[Kafka, Throwable, StringProducer] = + (producerSettings.toLayer ++ ZLayer.succeed(Serde.string: Serializer[Any, String])) >>> Producer.live[Any, String, String] - val testConfigLayer: URLayer[Clock with Blocking with Kafka, Clock with Blocking with ClustersConfigService] = + val testConfigLayer: URLayer[Clock with Kafka, Clock with ClustersConfigService] = ZLayer.requires[Clock] ++ - ZLayer.requires[Blocking] ++ ZLayer.fromService[Kafka.Service, ClustersConfig.Service] { kafka => new ClustersConfig.Service { def readClusters: Task[ClusterProperties] = @@ -39,7 +37,7 @@ trait HelperSpec { def produceMany( topic: String, kvs: Iterable[(String, String)] - ): RIO[Blocking with StringProducer, Chunk[RecordMetadata]] = + ): RIO[StringProducer, Chunk[RecordMetadata]] = Producer .produceChunk[Any, String, String](Chunk.fromIterable(kvs.map { case (k, v) => new ProducerRecord(topic, k, v) })) } diff --git a/core/src/test/scala/io/kafkamate/util/KafkaEmbedded.scala b/core/src/test/scala/io/kafkamate/util/KafkaEmbedded.scala index 3ea06bd..334609b 100644 --- a/core/src/test/scala/io/kafkamate/util/KafkaEmbedded.scala +++ b/core/src/test/scala/io/kafkamate/util/KafkaEmbedded.scala @@ -6,7 +6,7 @@ import net.manub.embeddedkafka.{EmbeddedK, EmbeddedKafka, EmbeddedKafkaConfig} import zio.* object KafkaEmbedded { - type Kafka = Has[Kafka.Service] + type Kafka = Kafka.Service object Kafka { diff --git a/fx/src/main/scala/io/km8/fx/Main.scala b/fx/src/main/scala/io/km8/fx/Main.scala index 7fa9d73..ba78694 100644 --- a/fx/src/main/scala/io/km8/fx/Main.scala +++ b/fx/src/main/scala/io/km8/fx/Main.scala @@ -1,9 +1,9 @@ package io.km8.fx import javafx.application.Platform -import scala.collection.mutable.{Queue => SQueue} + +import scala.collection.mutable.Queue as SQueue import zio.* -import zio.duration.* import javafx.beans.property.ObjectProperty import scalafx.application.JFXApp3 import scalafx.collections.ObservableBuffer @@ -15,8 +15,8 @@ import scalafx.scene.paint.* import scalafx.scene.text.* import io.km8.fx.models.* import io.km8.fx.models.given -import io.km8.fx.ui.{given, *} -import io.km8.fx.ui.components.{given, *} +import io.km8.fx.ui.{*, given} +import io.km8.fx.ui.components.{*, given} import java.util.concurrent.Executor import scala.concurrent.ExecutionContext @@ -24,7 +24,7 @@ import scalafx.scene.input.KeyEvent import scalafx.Includes.* import scalafx.scene.input.KeyCode -object Main extends JFXApp3 with BootstrapRuntime: +object Main extends JFXApp3: val currentThreadEC = ExecutionContext.fromExecutor(new Executor { override def execute(command: Runnable): Unit = command.run() @@ -35,12 +35,12 @@ object Main extends JFXApp3 with BootstrapRuntime: header <- HeaderControl().render navigator <- NavigatorControl().render mainContent <- MainContentControl().render - pane <- ZIO(new SplitPane { + pane <- ZIO.attempt(new SplitPane { dividerPositions = 0 id = "page-splitpane" items.addAll(navigator, mainContent) }) - p <- ZIO(new BorderPane { + p <- ZIO.attempt(new BorderPane { top = new VBox { vgrow = Priority.Always hgrow = Priority.Always @@ -52,7 +52,7 @@ object Main extends JFXApp3 with BootstrapRuntime: }) yield p - private def mkScene(sceneRoot: Parent): ZIO[Has[EventsHub], Throwable, Scene] = + private def mkScene(sceneRoot: Parent): ZIO[EventsHub, Throwable, Scene] = for dispatchFocusOmni <- dispatchEvent yield new Scene(1366, 768) { stylesheets = List("css/app.css") @@ -72,13 +72,14 @@ object Main extends JFXApp3 with BootstrapRuntime: hubLayer = ZLayer.succeed(h) main <- mkWindow.provideLayer(UI.make("") +!+ hubLayer) s <- mkScene(main).provideLayer(hubLayer) - ret <- ZIO(new JFXApp3.PrimaryStage { + ret <- ZIO.attempt(new JFXApp3.PrimaryStage { title = "KM8" scene = s }) yield ret - unsafeRun( + + Runtime.default.unsafeRun( io - .on(currentThreadEC) + .onExecutionContext(currentThreadEC) .exitCode ) diff --git a/fx/src/main/scala/io/km8/fx/models/UI.scala b/fx/src/main/scala/io/km8/fx/models/UI.scala index 21fb6e2..66921d9 100644 --- a/fx/src/main/scala/io/km8/fx/models/UI.scala +++ b/fx/src/main/scala/io/km8/fx/models/UI.scala @@ -2,10 +2,7 @@ package io.km8.fx package models import zio.* -import zio.duration.* -import zio.clock.* -import zio.prelude.* -import Assertion.* +import zio.prelude.NonEmptyList import scala.annotation.implicitNotFound @@ -60,18 +57,13 @@ def switch[P <: Page, S <: Page](p: Path[P, S]) = p match { } */ -inline def isNotEmpty: Assertion[String] = hasLength(greaterThan(0)) +//inline def isNotEmpty: Assertion[String] = hasLength(greaterThan(0)) -object Offset extends Subtype[Long] -type Offset = Offset.Type +opaque type Offset = Long -object Config extends Newtype[(String, String)] -type Config = Config.Type +opaque type Config = (String, String) -object TopicName extends Newtype[String]: - override inline def assertion = isNotEmpty - -type TopicName = TopicName.Type +opaque type TopicName = String /* Default type class @@ -80,7 +72,7 @@ trait Def[T]: def apply(seed: TestSeed): T given Def[TopicName] with - def apply(seed: TestSeed): TopicName = TopicName.make(s"topic $seed").getOrElse(TopicName("nothing")) + def apply(seed: TestSeed): TopicName = s"topic $seed" def gen[T: Def](seed: TestSeed = "test"): T = summon[Def[T]].apply(seed) @@ -108,7 +100,7 @@ case class Partition( size: Long) given Def[Partition] with - def apply(seed: TestSeed) = Partition(0, gen(seed), NonEmptyList(gen(seed)), Offset(0), Offset(0), 0) + def apply(seed: TestSeed) = Partition(0, gen(seed), NonEmptyList(gen(seed)), 0L, 0L, 0) case class Topic( name: TopicName, @@ -121,7 +113,7 @@ given Def[Topic] with def apply(seed: TestSeed) = Topic( - TopicName.make(s"topic $seed").getOrElse(TopicName("topic")), + s"topic $seed", NonEmptyList(gen(seed)), Nil, MessageEncoding.String, @@ -131,7 +123,7 @@ given Def[Topic] with case class PartitionOffset(partition: Partition, offset: Offset) given Def[PartitionOffset] with - def apply(seed: TestSeed) = PartitionOffset(gen(seed), Offset(0)) + def apply(seed: TestSeed) = PartitionOffset(gen(seed), 0L) case class ConsumerGroup( name: String, @@ -144,7 +136,7 @@ given Def[ConsumerGroup] with ConsumerGroup(s"name $seed", gen(seed), List.range(0, 10).map(gen)) given Def[Config] with - def apply(seed: TestSeed) = Config(s"key_$seed" -> s"value $seed") + def apply(seed: TestSeed): Config = s"key_$seed" -> s"value $seed" case class Cluster( name: String, @@ -180,17 +172,17 @@ enum UIEvent: type EventsHub = Hub[UIEvent] -val EventsHub = Hub +val EventsHub = zio.Hub object UI: - def make(seed: TestSeed = ""): URLayer[Any, Has[UI]] = - UIO( + def make(seed: TestSeed = ""): URLayer[Any, UI] = + ZLayer.succeed( UI( data = List.range(1, 4).map(gen), config = UIConfig(leftWidth = 300) ) - ).toLayer + ) case class Message( key: Array[Byte], @@ -199,7 +191,11 @@ case class Message( case class MessageHeader(key: String, value: Array[Byte]) -type UIEnv = Has[UI] & Has[EventsHub] +type UIEnv = UI & EventsHub -def dispatchEvent: ZIO[Has[EventsHub], Nothing, UIEvent => Unit] = +def dispatchEvent: ZIO[EventsHub, Nothing, UIEvent => Unit] = ZIO.service[EventsHub].map(hub => event => Runtime.global.unsafeRun(hub.publish(event))) + +object Test { + val stuff: String = "" +} \ No newline at end of file diff --git a/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala b/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala index 0addfde..cc36f45 100644 --- a/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala +++ b/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala @@ -6,7 +6,7 @@ import scalafx.scene.control.Alert.AlertType import scalafx.scene.control.{Alert, ButtonType} import zio.* -trait BaseControl[R <: Has[_]]: +trait BaseControl[R ]: def render: ZIO[R, Throwable, Node] def alert(text: Any) = diff --git a/fx/src/main/scala/io/km8/fx/ui/components/ClusterListControl.scala b/fx/src/main/scala/io/km8/fx/ui/components/ClusterListControl.scala index 2680ae8..d791a41 100644 --- a/fx/src/main/scala/io/km8/fx/ui/components/ClusterListControl.scala +++ b/fx/src/main/scala/io/km8/fx/ui/components/ClusterListControl.scala @@ -9,7 +9,7 @@ import scalafx.scene.Node import models.* -class ClusterListControl extends BaseControl[Has[UI]] { +class ClusterListControl extends BaseControl[UI] { val mkList = for { ui <- ZIO.service[UI] @@ -18,5 +18,5 @@ class ClusterListControl extends BaseControl[Has[UI]] { private[components] val view = mkList - override def render: RIO[Has[UI], Node] = view + override def render: RIO[UI, Node] = view } diff --git a/fx/src/main/scala/io/km8/fx/ui/components/HeaderControl.scala b/fx/src/main/scala/io/km8/fx/ui/components/HeaderControl.scala index c797b65..7257c57 100644 --- a/fx/src/main/scala/io/km8/fx/ui/components/HeaderControl.scala +++ b/fx/src/main/scala/io/km8/fx/ui/components/HeaderControl.scala @@ -32,7 +32,7 @@ class HeaderControl extends BaseControl[UIEnv]: _ <- ZStream .fromHub(hub) .foreach { case UIEvent.FocusOmni => - ZIO { + ZIO.succeed { omniBar.requestFocus() } } diff --git a/fx/src/main/scala/io/km8/fx/ui/components/MainContentControl.scala b/fx/src/main/scala/io/km8/fx/ui/components/MainContentControl.scala index 6162bd1..a404185 100644 --- a/fx/src/main/scala/io/km8/fx/ui/components/MainContentControl.scala +++ b/fx/src/main/scala/io/km8/fx/ui/components/MainContentControl.scala @@ -8,8 +8,8 @@ import zio.* import models.* -class MainContentControl extends BaseControl[Has[UI]]: +class MainContentControl extends BaseControl[UI]: - private lazy val view = ZIO(new ScrollPane()) + private lazy val view = ZIO.succeed(new ScrollPane()) - override def render: ZIO[Has[UI], Throwable, Node] = view + override def render: ZIO[UI, Throwable, Node] = view diff --git a/fx/src/main/scala/io/km8/fx/ui/components/NavigatorControl.scala b/fx/src/main/scala/io/km8/fx/ui/components/NavigatorControl.scala index f5c04c9..e28712f 100644 --- a/fx/src/main/scala/io/km8/fx/ui/components/NavigatorControl.scala +++ b/fx/src/main/scala/io/km8/fx/ui/components/NavigatorControl.scala @@ -7,13 +7,13 @@ import scalafx.scene.control.{ScrollPane, TreeItem, TreeView} import zio.* import models.* -class NavigatorControl extends BaseControl[Has[UI]]: +class NavigatorControl extends BaseControl[UI]: private lazy val view = for ui <- ZIO.service[UI] tv <- treeView - ret <- ZIO( + ret <- ZIO.succeed( new ScrollPane { minWidth = ui.config.leftWidth fitToWidth = true @@ -38,7 +38,7 @@ class NavigatorControl extends BaseControl[Has[UI]]: private val treeView = for ui <- ZIO.service[UI] - tv <- ZIO { + tv <- ZIO.succeed { new TreeView[String] { showRoot = false id = "left-tree" @@ -54,6 +54,6 @@ class NavigatorControl extends BaseControl[Has[UI]]: .addListener((obs, oldVal, newVal) => alert(ui.data.map(_.consumerGroups))) yield tv - override def render: ZIO[Has[UI], Throwable, Node] = + override def render: ZIO[UI, Throwable, Node] = for ret <- view yield ret diff --git a/fx/src/main/scala/io/km8/fx/views/Clusters.scala b/fx/src/main/scala/io/km8/fx/views/Clusters.scala new file mode 100644 index 0000000..d310716 --- /dev/null +++ b/fx/src/main/scala/io/km8/fx/views/Clusters.scala @@ -0,0 +1,3 @@ +package io.km8.fx.views + +case class Clusters(list: List[String]) \ No newline at end of file diff --git a/project/plugins.sbt b/project/plugins.sbt index b3d626e..8d769e4 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -7,3 +7,5 @@ addSbtPlugin("org.jmotor.sbt" % "sbt-dependency-updates" % "1.2.2") addSbtPlugin("se.marcuslonnberg" % "sbt-docker" % "1.9.0") addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "1.2.0") addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.6") +// project/plugins.sbt +addSbtPlugin("ch.epfl.scala" % "sbt-scalafix" % "0.10.1") \ No newline at end of file