kafka pipe working!!!!!!!!!!!!!!!!!

lots of experiments too - to be deleted

buildSrc/src/main/kotlin/dev/adamko/factoriowebmap/archetype/kotlin-jvm.gradle.kts

@@ -8,6 +8,9 @@ plugins {
 }
 
 kotlin {
+  jvmToolchain {
+    (this as JavaToolchainSpec).languageVersion.set(JavaLanguageVersion.of(11)) // "8"
+  }
 }
 
 dependencies {
@@ -16,8 +19,9 @@ dependencies {
 
 tasks.withType<KotlinCompile>().configureEach {
   kotlinOptions.freeCompilerArgs += listOf(
-      "-Xopt-in=kotlin.OptIn",
-      "-Xopt-in=kotlin.ExperimentalStdlibApi",
-      "-Xopt-in=kotlin.time.ExperimentalTime",
+    "-Xopt-in=kotlin.OptIn",
+    "-Xopt-in=kotlin.ExperimentalStdlibApi",
+    "-Xopt-in=kotlin.time.ExperimentalTime",
+    "-Xopt-in=kotlinx.coroutines.ExperimentalCoroutinesApi",
   )
 }

modules/factorio-events-kafka-pipe/build.gradle.kts

@@ -0,0 +1,155 @@
+import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs
+
+plugins {
+  id("dev.adamko.factoriowebmap.archetype.kotlin-jvm")
+
+  application
+
+//  id("com.github.johnrengelman.shadow") version "7.1.0"
+}
+
+description =
+  """
+    Send events from a Factorio server to a Kafka topic 
+  """.trimIndent()
+
+val projectId: String by project.extra
+
+dependencies {
+  implementation(libs.bundles.logging)
+  implementation("eu.jrie.jetbrains:kotlin-shell-core:0.2.1")
+}
+
+application {
+  mainClass.set("factorioEventsKafkaPipeKt")
+}
+
+java {
+  withSourcesJar()
+}
+
+tasks.jar {
+  manifest {
+    attributes(
+      "Implementation-Title" to project.name,
+      "Implementation-Version" to project.version,
+      "Main-Class" to application.mainClass,
+    )
+  }
+  archiveBaseName.set(project.name)
+  duplicatesStrategy = DuplicatesStrategy.EXCLUDE
+
+  from(
+    configurations.runtimeClasspath.map { config ->
+      config.map {
+        logger.trace("Adding runtimeClasspath file to jar - $it")
+        when (it.isDirectory) {
+          true -> it
+          false -> zipTree(it)
+        }
+      }
+    }
+  )
+
+}
+
+tasks.create<Exec>("dockerBuildPdl2k") {
+  group = projectId
+
+  dependsOn(tasks.jar)
+
+  val jar = tasks.jar.get().archiveFile.get()
+  logger.lifecycle("building docker with jar $jar")
+  inputs.file(jar)
+
+  commandLine(
+    parseSpaceSeparatedArgs(
+      "docker-compose build pdl2k --build-arg JAR_LOCATION=${jar.asFile.canonicalPath}"
+    )
+  )
+
+//  inputs(tasks.jar.get().archiveFile.get())
+
+//  commandLine(
+//    "docker",
+//    "build",
+//    "--build-arg",
+//    "JAR_FILE=",
+//    "-t dev-adamko/pdl2k"
+//  )
+}
+
+//tasks.jar.get().archiveFile.get().asFile.canonicalPath
+
+tasks.create<Exec>("runInDocker") {
+  group = projectId
+//   docker run -v `pwd`:/mnt java:8 java -jar /mnt/myapp.jar
+
+  dependsOn(tasks.jar)
+  val jarFile = tasks.jar.get().archiveFile.get().asFile
+  inputs.file(jarFile)
+
+  commandLine(
+    listOf(
+      "docker",
+      "run",
+      "--rm",
+      "--name", project.name,
+//      """--workdir="/app" """,
+      "-v", ".//${jarFile.toRelativeString(project.projectDir).replace("\\", "//")}:/mnt/",
+      "openjdk:11.0.6-jre-buster",
+      "ls", "-la", "/mnt/*",
+//      "java", "-jar", "/mnt/${jarFile.name}",
+    )
+  )
+
+  doFirst {
+    logger.lifecycle("Executing [$commandLine]")
+  }
+}
+
+//afterEvaluate {
+//  tasks {
+//    create("jar2", Jar::class).apply {
+//      dependsOn("jvmMainClasses", "jsJar")
+//      group = "jar"
+//      manifest {
+//        attributes(
+//          mapOf(
+//            "Implementation-Title" to rootProject.name,
+//            "Implementation-Version" to rootProject.version,
+//            "Timestamp" to System.currentTimeMillis(),
+//            "Main-Class" to application.mainClass
+//          )
+//        )
+//      }
+//
+//      from(
+//        configurations
+//      )
+//
+//      val dependencies =
+//        configurations["jvmRuntimeClasspath"]
+//          .filter { it.name.endsWith(".jar") } +
+//            project.tasks["jvmJar"].outputs.files
+//      dependencies.forEach { from(zipTree(it)) }
+//      into("/lib")
+//    }
+//  }
+//}
+//val fatJar = task("fatJar", type = Jar::class) {
+//  baseName = "safescape-lib-${project.name}"
+//  // manifest Main-Class attribute is optional.
+//  // (Used only to provide default main class for executable jar)
+//  from(configurations.runtimeClasspath.map({ if (it.isDirectory) it else zipTree(it) }))
+//  with(tasks["jar"] as CopySpec)
+//}
+//
+//tasks {
+//  "build" {
+//    dependsOn(fatJar)
+//  }
+//}
+//from(configurations.runtimeClasspath) {
+//  into("lib")
+//}

modules/factorio-events-kafka-pipe/docker-compose.yml

@@ -0,0 +1,54 @@
+version: "3.9"
+
+services:
+  kp2:
+    image: openjdk:11-jre-buster
+    container_name: "factorio-events-kafka-pipe"
+    build:
+      dockerfile: "kafka-pipe.Dockerfile"
+    networks:
+      - "factorio-events-nw"
+    environment:
+      FACTORIO_SERVER_CONTAINER_NAME: "factorio-server"
+      KAFKA_HOST: "kafka:29092"
+    volumes:
+      # Unix and Windows?
+      #      - "//./pipe/docker_engine:/var/run/docker.sock"
+      #      - "//./pipe/docker_engine:/pipe/docker_engine"
+      - "//var/run/docker.sock:/var/run/docker.sock"
+      #      - "/var/run/docker.sock:/var/run/docker.sock"
+      # Windows?
+      #      - "\\.\\pipe\\docker_engine:\\.\\pipe\\docker_engine"
+      # Windows?
+      #      - "/usr/local/bin/docker:/usr/bin/docker"
+
+
+#  kp3:
+#    image: edenhill/kcat:1.7.0
+#    container_name: kafka-pipe3
+#    networks:
+#      - "factorio-events-nw"
+#    volumes:
+#      - "./kafka-pipe.sh:/kafka-pipe.sh"
+#    entrypoint:
+#      #      - "ls"
+#      #      - "-la"
+#      #      - "&&"
+#      - "/bin/sh"
+#      - "./kafka-pipe.sh"
+#      - "factorio-server"
+#      - "kafka-kraft"
+
+###
+
+#    volumes:
+#      - "./build/libs/:/app"
+#    command:
+#      - "java"
+#      - "-jar"
+#      - "/app/pipe-docker-logs-to-kafka-0.0.2.jar"
+
+networks:
+  factorio-events-nw:
+    name: "factorio-events-nw"
+    external: true

modules/factorio-events-kafka-pipe/kafka-pipe-jar.Dockerfile

@@ -0,0 +1,12 @@
+ARG JAR_LOCATION
+
+FROM openjdk:11.0.6-jre-buster
+ARG JAR_LOCATION
+
+RUN apt-get update && apt-get install -y \
+  curl \
+  && rm -rf /var/lib/apt/lists/*
+
+COPY JAR_LOCATION /app.jar
+
+ENTRYPOINT ["java","-jar","/app.jar"]

modules/factorio-events-kafka-pipe/kafka-pipe.Dockerfile

@@ -0,0 +1,31 @@
+
+#FROM debian:buster-slim
+FROM docker:dind-rootless
+
+ENV FACTORIO_SERVER_CONTAINER_NAME="factorio-server"
+ENV KAFKA_HOST="kafka-kraft"
+
+#RUN apt-get update && apt-get install -y --no-install-recommends \
+#  curl \
+#  docker \
+#  kafkacat
+
+# root is required for following Docker logs
+USER root
+
+RUN apk add --update \
+    curl \
+    kafkacat
+#USER rootless
+
+#     \
+#  && rm -rf /var/lib/apt/lists/*
+
+COPY kafka-pipe.sh kafka-pipe.sh
+
+#RUN useradd --user-group --system --create-home --no-log-init the_engineer
+#USER the_engineer
+
+ENTRYPOINT [ "/bin/sh", "/kafka-pipe.sh" ]
+#ENTRYPOINT /bin/bash
+#CMD exec /kafka-pipe.sh

modules/factorio-events-kafka-pipe/kafka-pipe.sh

@@ -0,0 +1,21 @@
+#!/usr/bin/bash
+
+# Pipe Docker logs from a Factorio server to a Kafka topic
+
+echo "Piping logs from $FACTORIO_SERVER_CONTAINER_NAME to $KAFKA_HOST"
+
+trap endProcess SIGINT
+endProcess() {
+  echo "Exiting Factorio Events Kafka Pipe"
+  exit
+}
+
+while :; do
+  docker logs --tail 0 -f "$FACTORIO_SERVER_CONTAINER_NAME" |
+    sed -n -e 's/^FactorioEvent: //p' |
+    kafkacat -P -b "$KAFKA_HOST" -t factorio-server-log
+
+  echo "Error - retrying in 10 seconds"
+
+  sleep 10
+done

modules/factorio-events-kafka-pipe/pipe-logs-to-kafka.sh

@@ -0,0 +1,50 @@
+#!/usr/bin/zsh
+
+# stream logs from 'factorio' container
+# follow the logs, and
+# combine stdout and stderr
+# only match lines that start with 'FactorioEvent: ',
+# and only return the stuff *after* 'FactorioEvent: '
+docker logs --tail 0 -f factorio-server 2>&1 |
+  sed -n -e 's/^FactorioEvent: //p' |
+  kafkacat -P -b localhost -t factorio-server-log
+
+# listen to events
+# kafkacat -C -b localhost -t factorio-server-log | jq
+
+# curl -sS --unix-socket /var/run/docker.sock "http://localhost/v1.41/events?filters={"service":"factorio-server"}" --output - | jq
+
+curl --no-buffer -sS \
+  --unix-socket /var/run/docker.sock \
+  "http://localhost/v1.41/containers/factorio-server/logs?stdout=true&follow=true&tail=0"  |
+   tail -c+9
+#  --output -
+
+curl \
+  -sS \
+  --unix-socket /var/run/docker.sock \
+  -X POST \
+  --insecure "http://localhost/v1.41/containers/factorio-server/attach?logs=true&stream=true&stdout=true&follow=true" |
+  tail -c+9
+
+#  sed 's/^........//'
+
+
+# https://stackoverflow.com/questions/42873285/curl-retry-mechanism
+# --retry-all-errors
+
+#     --header "Sec-WebSocket-Key: SGVsbG8sIHdvcmxkIQ==" \
+#     --header "Sec-WebSocket-Version: 13" \
+#     --header "Host: localhost:80" \
+#     --header "Origin: http://localhost:80" \
+
+curl --include \
+     --unix-socket /var/run/docker.sock \
+     --no-buffer \
+     --header "Connection: Upgrade" \
+     --header "Upgrade: websocket" \
+     --header "Host: localhost:80" \
+     --header "Origin: http://localhost:80" \
+     --header "Sec-WebSocket-Version: 13" \
+     --header "Sec-WebSocket-Key: SGVsbG8sIHdvcmxkIQ==" \
+     "http://localhost/v1.41/containers/factorio-server/attach/ws?logs=true&stdout=true"

modules/factorio-events-kafka-pipe/src/main/kotlin/fekp.kt

@@ -0,0 +1,17 @@
+import eu.jrie.jetbrains.kotlinshell.shell.shell
+
+fun main() {
+  shell {
+
+    "echo hello world"()
+
+//    """
+//      curl \
+//        --no-buffer \
+//        -sS \
+//        --unix-socket /var/run/docker.sock \
+//        "http://localhost/v1.41/containers/factorio-server/logs?stdout=true&follow=true&tail=0"
+//    """.trimIndent()
+
+  }
+}

modules/factorio-events-processor/build.gradle.kts

@@ -12,4 +12,3 @@ description =
 val projectId: String by project.extra
 val buildDir: Directory = layout.buildDirectory.dir(projectId).get()
 
-

modules/mod/infra/docker-compose.yml

@@ -8,6 +8,8 @@ services:
     #      driver: "json-file"
     container_name: "factorio-server"
     restart: "unless-stopped"
+    networks:
+      - "factorio-events-nw"
     ports:
       - "34197:34197/udp" # factorio
       - "27015:27015/tcp" # rcon
@@ -29,18 +31,28 @@ services:
   #      - USERNAME=FactorioUsername
   #      - TOKEN=FactorioToken
 
-  kafka-kraft:
+  kafka:
     image: dev-adamko/kafka-kraft
-    container_name: "kafka-kraft"
+    container_name: "kafka"
+    restart: "unless-stopped"
     build:
       dockerfile: "kafka.Dockerfile"
       target: "kafka-kraft"
       args:
         KAFKA_DL_URL: "https://dlcdn.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz"
+    networks:
+      - "factorio-events-nw"
     ports:
       - "9092:9092"
       - "9093:9093"
 
+#    environment:
+#      KAFKA_LISTENERS: "PLAINTEXT://kafka-kraft:29092,PLAINTEXT_HOST://localhost:9092,CONTROLLER_HOST://localhost:9093" # ,CONTROLLER://kafka-kraft:29093
+#      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT" #,CONTROLLER_HOST:PLAINTEXT
+#      KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
+#      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-kraft:29092,PLAINTEXT_HOST://localhost:9092"
+#      KAKFA_CONTROLLER_LISTENER_NAMES: "CONTROLLER" #,CONTROLLER_HOST
+
 #  kafka-connect:
 #    image: dev-adamko/kafka-connect
 #    container_name: "kafka-connect"
@@ -54,3 +66,9 @@ services:
 #    ports:
 #      - "8083:8083"
 #
+
+
+networks:
+  factorio-events-nw:
+    name: "factorio-events-nw"
+

modules/mod/infra/kafka-server.properties

@@ -0,0 +1,135 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#
+# This configuration file is intended for use in KRaft mode, where
+# Apache ZooKeeper is not present.  See config/kraft/README.md for details.
+#
+
+############################# Server Basics #############################
+
+# The role of this server. Setting this puts us in KRaft mode
+process.roles=broker,controller
+
+# The node id associated with this instance's roles
+node.id=1
+
+# The connect string for the controller quorum
+controller.quorum.voters=1@localhost:9093
+
+############################# Socket Server Settings #############################
+
+# The address the socket server listens on. It will get the value returned from
+# java.net.InetAddress.getCanonicalHostName() if not configured.
+#   FORMAT:
+#     listeners = listener_name://host_name:port
+#   EXAMPLE:
+#     listeners = PLAINTEXT://your.host.name:9092
+#listeners=PLAINTEXT://:9092,CONTROLLER://:9093
+#listeners=PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092,CONTROLLER://localhost:9093
+listeners=PLAINTEXT://:29092,PLAINTEXT_HOST://:9092,CONTROLLER://:9093
+
+inter.broker.listener.name=PLAINTEXT
+#KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
+
+# Hostname and port the broker will advertise to producers and consumers. If not set,
+# it uses the value for "listeners" if configured.  Otherwise, it will use the value
+# returned from java.net.InetAddress.getCanonicalHostName().
+#advertised.listeners=PLAINTEXT://localhost:9092
+advertised.listeners=PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
+
+# Listener, host name, and port for the controller to advertise to the brokers. If
+# this server is a controller, this listener must be configured.
+controller.listener.names=CONTROLLER
+
+# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
+#listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
+listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+#  ,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
+
+# The number of threads that the server uses for receiving requests from the network and sending responses to the network
+num.network.threads=3
+
+# The number of threads that the server uses for processing requests, which may include disk I/O
+num.io.threads=8
+
+# The send buffer (SO_SNDBUF) used by the socket server
+socket.send.buffer.bytes=102400
+
+# The receive buffer (SO_RCVBUF) used by the socket server
+socket.receive.buffer.bytes=102400
+
+# The maximum size of a request that the socket server will accept (protection against OOM)
+socket.request.max.bytes=104857600
+
+
+############################# Log Basics #############################
+
+# A comma separated list of directories under which to store log files
+log.dirs=/tmp/kraft-combined-logs
+
+# The default number of log partitions per topic. More partitions allow greater
+# parallelism for consumption, but this will also result in more files across
+# the brokers.
+num.partitions=1
+
+# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
+# This value is recommended to be increased for installations with data dirs located in RAID array.
+num.recovery.threads.per.data.dir=1
+
+############################# Internal Topic Settings  #############################
+# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
+# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
+offsets.topic.replication.factor=1
+transaction.state.log.replication.factor=1
+transaction.state.log.min.isr=1
+
+############################# Log Flush Policy #############################
+
+# Messages are immediately written to the filesystem but by default we only fsync() to sync
+# the OS cache lazily. The following configurations control the flush of data to disk.
+# There are a few important trade-offs here:
+#    1. Durability: Unflushed data may be lost if you are not using replication.
+#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
+#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
+# The settings below allow one to configure the flush policy to flush data after a period of time or
+# every N messages (or both). This can be done globally and overridden on a per-topic basis.
+
+# The number of messages to accept before forcing a flush of data to disk
+#log.flush.interval.messages=10000
+
+# The maximum amount of time a message can sit in a log before we force a flush
+#log.flush.interval.ms=1000
+
+############################# Log Retention Policy #############################
+
+# The following configurations control the disposal of log segments. The policy can
+# be set to delete segments after a period of time, or after a given size has accumulated.
+# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
+# from the end of the log.
+
+# The minimum age of a log file to be eligible for deletion due to age
+log.retention.hours=168
+
+# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
+# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
+#log.retention.bytes=1073741824
+
+# The maximum size of a log segment file. When this size is reached a new log segment will be created.
+log.segment.bytes=1073741824
+
+# The interval at which log segments are checked to see if they can be deleted according
+# to the retention policies
+log.retention.check.interval.ms=300000

modules/mod/infra/kafka.Dockerfile

@@ -41,8 +41,10 @@ RUN ./bin/kafka-storage.sh format \
   --config ./config/kraft/server.properties \
   --cluster-id "$(cat cluster_id)"
 
+COPY ./kafka-server.properties ./server.properties
+
 # launch the broker in KRaft mode, which means that it runs without ZooKeeper
-ENTRYPOINT ["./bin/kafka-server-start.sh", "./config/kraft/server.properties"]
+ENTRYPOINT ["./bin/kafka-server-start.sh", "./server.properties"]
 
 
 ### Kafka Connect ##

settings.gradle.kts

@@ -7,6 +7,7 @@ include(
     ":modules:kt-rcon",
     ":modules:factorio-events-data-model",
     ":modules:factorio-events-processor",
+    ":modules:factorio-events-kafka-pipe",
 )