Initial implementation

build.gradle

@@ -2,11 +2,11 @@ plugins {
     id 'application'
     id 'distribution'
     id 'org.jetbrains.kotlin.jvm' version '1.3.71'
-    id 'com.gitlab.morality.grit' version '2.0.2'
     id 'org.jlleitschuh.gradle.ktlint' version '9.2.1'
 }
 
 group 'org.memobase'
+version '1.0'
 mainClassName = 'org.memobase.App'
 
 jar {

src/main/kotlin/App.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.memobase
+
+import kotlin.system.exitProcess
+import org.apache.logging.log4j.LogManager
+
+class App {
+    companion object {
+        private val log = LogManager.getLogger("GroupReportsConsumerApp")
+        @JvmStatic fun main(args: Array<String>) {
+            try {
+                Service().run()
+            } catch (ex: Exception) {
+                ex.printStackTrace()
+                log.error("Stopping application due to error: " + ex.message)
+                exitProcess(1)
+            }
+        }
+    }
+}

src/main/kotlin/Consumer.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.memobase
+
+import org.apache.kafka.clients.consumer.ConsumerRecords
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import java.time.Duration
+import java.util.*
+
+class Consumer(topic: String, properties: Properties) {
+    private val instance = KafkaConsumer<String, String>(properties)
+
+    init {
+        instance.subscribe(listOf(topic))
+    }
+
+    fun consume(): ConsumerRecords<String, String> {
+        return instance.poll(Duration.ofSeconds(10L))
+    }
+}
\ No newline at end of file

src/main/kotlin/IndexReport.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.memobase
+
+import ch.memobase.reporting.Report
+import ch.memobase.settings.HeaderMetadata
+
+class IndexReport(
+    val report: Report,
+    val metadata: HeaderMetadata
+)
\ No newline at end of file

src/main/kotlin/ProcessReport.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.memobase
+
+data class ProcessReport(
+    val identifier: String,
+    val type: Type,
+    val fatal: Boolean,
+    val stepsCompleted: Int,
+    val message: String
+) {
+    fun toRequestBody(): RequestBody {
+        return RequestBody(
+            type.name,
+            identifier,
+            !fatal,
+            message
+        )
+    }
+}

src/main/kotlin/Producer.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.memobase
+
+import ch.memobase.reporting.Report
+import java.io.Closeable
+import java.util.Properties
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.header.Headers
+
+class Producer(
+    private val reportingTopic: String,
+    producerProps: Properties
+) : Closeable {
+    private val instance = KafkaProducer<String, String>(producerProps)
+    fun sendReport(report: Report, headers: Headers?) {
+        instance.send(ProducerRecord(reportingTopic, null, report.id, report.toJson(), headers))
+    }
+    override fun close() {
+        instance.close()
+    }
+}

src/main/kotlin/ReportCollector.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.memobase
+
+import ch.memobase.reporting.Report
+import ch.memobase.reporting.ReportStatus
+import com.beust.klaxon.Klaxon
+import org.apache.logging.log4j.LogManager
+import java.net.HttpURLConnection
+import java.net.URL
+
+class ReportCollector(private val importApiUrl: String) {
+    private val log = LogManager.getLogger(this::class.java)
+    private val collection = mutableMapOf<String, ProcessReport>()
+    private val requiredStepsForInstitutions = 4
+    private val requiredStepsForRecordSets = 5
+    private val klaxon = Klaxon()
+
+    private val url = URL(importApiUrl)
+
+    fun addReport(identifier: String, type: Type, session: String, report: Report) {
+        val key = identifier + session
+        if (collection.containsKey(key)) {
+            val value = collection[key]!!
+            collection[key] = ProcessReport(
+                identifier,
+                type,
+                report.status == ReportStatus.fatal,
+                value.stepsCompleted + 1,
+                "${value.message}\n${report.message}"
+            )
+        } else {
+            collection[identifier + session] = ProcessReport(
+                identifier,
+                type,
+                report.status == ReportStatus.fatal,
+                1,
+                report.message
+            )
+        }
+    }
+
+    fun check() {
+        collection
+            .filter { it.value.fatal || hasRequired(it.value.stepsCompleted, it.value.type) }
+            .map { it.value.toRequestBody().toJson(klaxon) }
+            .forEach {
+                val con = url.openConnection()
+                val http = con as HttpURLConnection
+                http.requestMethod = "POST"
+                http.doOutput = true
+                http.setFixedLengthStreamingMode(it.size)
+                http.setRequestProperty("Content-Type", "application/json; charset=UTF-8")
+                http.connect()
+                http.outputStream.use { os -> os.write(it) }
+                if (http.responseCode >= 400) {
+                    http.errorStream.use { errorStream ->
+                        log.error(errorStream.bufferedReader().lines().reduce { s: String?, s2: String? -> s + s2 })
+                    }
+                }
+                http.disconnect()
+            }
+    }
+
+    private fun hasRequired(completed: Int, type: Type): Boolean {
+        return when (type) {
+            Type.institution -> completed >= requiredStepsForInstitutions
+            Type.record_set -> completed >= requiredStepsForRecordSets
+        }
+    }
+}
\ No newline at end of file

src/main/kotlin/RequestBody.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.memobase
+
+import com.beust.klaxon.Klaxon
+import org.apache.logging.log4j.LogManager
+
+data class RequestBody(
+    val node_type: String,
+    val id: String,
+    val status: Boolean,
+    val report: String
+) {
+    private val log = LogManager.getLogger(this::class.java)
+
+    fun toJson(klaxon: Klaxon): ByteArray {
+        val message = klaxon.toJsonString(this)
+        log.debug(message)
+        return message.toByteArray()
+    }
+}

src/main/kotlin/Service.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.memobase
+
+import ch.memobase.settings.SettingsLoader
+import com.beust.klaxon.Klaxon
+import org.apache.logging.log4j.LogManager
+
+class Service(file: String = "app.yml") {
+    companion object {
+        const val importApiUrlSettingName = "importApiUrl"
+        const val whitelistFilePathSettingName = "whitelistFilePath"
+    }
+
+    private val log = LogManager.getLogger("GroupReportsConsumerService")
+
+    private val settings = SettingsLoader(
+        listOf(
+            importApiUrlSettingName,
+            whitelistFilePathSettingName
+        ),
+        file,
+        useConsumerConfig = true,
+        useProducerConfig = true
+    )
+
+    private val klaxon = Klaxon()
+    private val consumer = Consumer(settings.inputTopic, settings.kafkaConsumerSettings)
+    private val producer = Producer(settings.outputTopic, settings.kafkaProducerSettings)
+    private val importApiUrl = settings.appSettings[importApiUrlSettingName] as String
+    private val stepWhitelist = StepWhitelist(settings.appSettings[whitelistFilePathSettingName] as String)
+
+    val collector = ReportCollector(importApiUrl)
+
+    fun run() {
+        while (true) {
+            consumer.consume()
+                .mapNotNull { klaxon.parse<IndexReport>(it.value()) }
+                .filter { stepWhitelist.check(it.report.step) }
+                .forEach {
+                    if (it.metadata.institutionId == "none") {
+                        collector.addReport(
+                            it.metadata.recordSetId,
+                            Type.record_set,
+                            it.metadata.sessionId,
+                            it.report
+                        )
+                    } else {
+                        collector.addReport(
+                            it.metadata.institutionId,
+                            Type.institution,
+                            it.metadata.sessionId,
+                            it.report
+                        )
+                    }
+
+                }
+            collector.check()
+        }
+    }
+}
+

src/main/kotlin/StepWhitelist.kt

+package org.memobase
+
+import java.io.File
+import java.nio.charset.Charset
+
+class StepWhitelist(path: String) {
+
+    private val whitelist = File(path).readLines(Charset.defaultCharset())
+
+    fun check(step: String): Boolean {
+        return whitelist.contains(step)
+    }
+}
\ No newline at end of file

src/main/kotlin/Type.kt

+package org.memobase
+
+enum class Type {
+    institution,
+    record_set
+}
\ No newline at end of file

src/test/kotlin/TestReportCollector.kt

+/*
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+import ch.memobase.reporting.Report
+import ch.memobase.reporting.ReportStatus
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.TestInstance
+import org.memobase.ReportCollector
+import org.memobase.Type
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class TestReportCollector {
+    private val stageURL = "https://stage.import.memobase.k8s.unibas.ch/v1/drupal/WriteElementReport"
+
+    @Test
+    fun `test collection`() {
+
+
+        val collector = ReportCollector(stageURL)
+
+        collector.addReport(
+            "csa",
+            Type.institution,
+            "1",
+            Report(
+                "csa",
+                ReportStatus.fatal,
+                "Test message new 1",
+                "gi-drupal-syncer"
+            )
+        )
+
+        collector.check()
+    }
+}
\ No newline at end of file

src/test/resources/app.yml

+app:
+  importApiUrl: https://stage.import.memobase.k8s.unibas.ch/v1/drupal/WriteElementReport
+kafka:
+  consumer:
+    bootstrap.servers: kafka:9092
+    client.id: client
+    group.id: client
+  producer:
+    bootstrap.servers: kafka:9092
+    client.id: client
+  topic:
+    in: in
+    out: out
+    process: process
\ No newline at end of file

src/test/resources/log4j2.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="INFO">
+    <Appenders>
+        <Console name="console" target="SYSTEM_OUT">
+            <PatternLayout
+                    pattern="[%-5level] [%c{1}] - %msg%n" />
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Root level="debug" additivity="false">
+            <AppenderRef ref="console" />
+        </Root>
+    </Loggers>
+</Configuration>
\ No newline at end of file