Skip to content

GitLab

Commit ebd4cfb7 authored by Jonas Waeber's avatar Jonas Waeber
Browse files

Refactor input

parent dbf8fa11
Hide whitespace changes
Inline Side-by-side
Showing with 280 additions and 252 deletions
helm-charts/templates/app-config.yaml
View file @ ebd4cfb7
apiVersion: v1
kind: ConfigMap
metadata:
name: "{{ .Values.processId }}-{{ .Values.jobName }}-config"
name: "{{ .Values.recordSetId }}-{{ .Values.shortSessionId }}-config"
namespace: memobase
data:
APP_DIRECTORY: "{{ .Values.appDirectory }}"
CLIENT_ID: "{{ .Values.processId }}-{{ .Values.jobName }}"
TOPIC_IN: "{{ .Values.processId }}-{{ .Values.jobName }}"
TOPIC_PROCESS: "{{ .Values.processId }}-reporting"
\ No newline at end of file
SESSION_ID: "{{ .Values.sessionId }}"
RECORD_SET_ID: "{{ .Values.recordSetId }}"
INSTITUTION_ID: "{{ .Values.institutionId }}"
IS_PUBLISHED: "{{ .Values.isPublished }}"
XML_RECORD_TAG: "{{.Values.xmlRecordTag }}"
XML_IDENTIIFER_FIELD_NAME: "{{ .Values.xmlIdentifierFieldName }}"
TABLE_SHEET_INDEX: "{{.Values.tableSheetIndex }}"
TABLE_HEADER_COUNT: "{{ .Values.tableHeaderCount }}"
TABLE_HEADER_INDEX: "{{ .Values.tableHeaderIndex }}"
TABLE_IDENTIFIER_INDEX: "{{ .Values.tableIdentifierIndex }}"
CLIENT_ID: "{{ .Values.recordSetId }}-{{ .Values.sessionId }}"
TOPIC_OUT: "{{.Values.topicName }}"
TOPIC_REPORTING: "{{ .Values.reportingTopicName }}"
\ No newline at end of file
helm-charts/templates/job.yaml
View file @ ebd4cfb7
apiVersion: batch/v1
kind: Job
metadata:
name: "{{ .Values.processId }}-{{ .Values.jobName }}"
name: "{{ .Values.reportingTopicName }}-{{ .Values.shortSessionId }}"
namespace: memobase
labels:
institutionId: "{{ .Values.institutionId }}"
recordSetId: "{{ .Values.recordSetId }}"
jobType: "import-job"
jobType: "text-file-validation"
spec:
template:
spec:
containers:
- name: "{{ .Values.processId }}-{{ .Values.jobName }}"
- name: "{{ .Values.recordSetId }}-{{ .Values.sessionId }}"
image: "{{ .Values.registry }}/{{ .Values.image }}:{{ .Values.tag }}"
envFrom:
- secretRef:
......@@ -19,6 +19,6 @@ spec:
- configMapRef:
name: "{{ .Values.kafkaConfigs }}"
- configMapRef:
name: "{{ .Values.processId }}-{{ .Values.jobName }}-config"
name: "{{ .Values.recordSetId }}-{{ .Values.shortSessionId }}-config"
restartPolicy: Never
backoffLimit: 0
\ No newline at end of file
helm-charts/values.yaml
View file @ ebd4cfb7
......@@ -9,20 +9,27 @@ tag: "latest"
kafkaConfigs: prod-kafka-bootstrap-servers
sftpConfigs: internal-sftp-config
topicName: import-process-data-transform
reportingTopicName: import-process-reporting
############################################
## Values below should be defined via the #
## User Interface (Drupal) #
############################################
jobName: text-file-validation
processId: p0001
###
# API Configs (Mandatory!)
###
sessionId: placeholder
shortSessionId: placeholder
institutionId: placeholder
recordSetId: placeholder
## Needs to be set to the directory on the sftp server.
## this is a relative path built like this:
## "./{RECORD_SET_ID}"
## The exact structure will be defined in task MEMO-196
appDirectory: placeholderValue
\ No newline at end of file
isPublished: false
# Step 2
# xml-data-transform
xmlRecordTag: record
xmlIdentifierFieldName: id
# table-data-transform
tableSheetIndex: 1
tableHeaderCount: 1
tableHeaderIndex: 1
tableIdentifierIndex: 1
src/main/kotlin/App.kt
View file @ ebd4cfb7
......@@ -28,6 +28,7 @@ class App {
try {
val service = Service()
service.run()
exitProcess(0)
} catch (ex: Exception) {
ex.printStackTrace()
log.error("Stopping application due to error: " + ex.message)
......
src/main/kotlin/ProcessReport.kt deleted 100644 → 0
View file @ dbf8fa11
/*
* xml-data-transform
* 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
data class ProcessReport(
val id: String,
val status: String,
val total: Int,
val successes: Int,
val failures: Int
) {
fun toJson(): String {
return Klaxon().toJsonString(this)
}
}
src/main/kotlin/Producer.kt
View file @ ebd4cfb7
......@@ -21,23 +21,31 @@ 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.Header
import org.apache.kafka.common.header.internals.RecordHeader
class Producer(props: Properties, private val topic: String) : Closeable {
class Producer(
props: Properties,
headers: Properties,
private val outputTopic: String,
private val reportingTopic: String
) : Closeable {
private val instance = KafkaProducer<String, String>(props)
private val reportingTopic = "$topic-reporting"
private val headers = transformHeaders(headers)
private fun transformHeaders(headers: Properties): List<Header> {
return headers.map { item ->
RecordHeader(item.key as String, (item.value as String).toByteArray())
}
}
fun sendMessage(key: String, message: Message) {
instance.send(ProducerRecord(topic, key, message.toJson()))
instance.send(ProducerRecord(outputTopic, null, key, message.toJson(), headers))
}
fun sendReport(report: Report) {
instance.send(ProducerRecord(reportingTopic, report.id, report.toJson()))
}
fun sendJobReport(report: ProcessReport, topic: String) {
instance.send(ProducerRecord(topic, report.id, report.toJson()))
}
override fun close() {
instance.close()
}
......
src/main/kotlin/Report.kt
View file @ ebd4cfb7
......@@ -19,13 +19,38 @@
package org.memobase
import com.beust.klaxon.Klaxon
import java.time.LocalDateTime
data class Report(
val id: String,
val step: String = "text-file-validation",
val timestamp: String = LocalDateTime.now().toString(),
val status: String,
val message: String
) {
fun toJson(): String {
return Klaxon().toJsonString(this)
}
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (javaClass != other?.javaClass) return false
other as Report
if (id != other.id) return false
if (step != other.step) return false
if (status != other.status) return false
if (message != other.message) return false
return true
}
override fun hashCode(): Int {
var result = id.hashCode()
result = 31 * result + step.hashCode()
result = 31 * result + status.hashCode()
result = 31 * result + message.hashCode()
return result
}
}
src/main/kotlin/Service.kt
View file @ ebd4cfb7
......@@ -31,14 +31,26 @@ class Service(fileName: String = "app.yml") {
val settings =
SettingsLoader(
listOf("directory"),
listOf(
"sessionId",
"recordSetId",
"institutionId",
"isPublished",
"xmlRecordTag",
"xmlIdentifierFieldName",
"tableSheetIndex",
"tableHeaderCount",
"tableHeaderIndex",
"tableIdentifierIndex"
),
fileName,
useProducerConfig = true,
readSftpSettings = true
)
private val producer = Producer(settings.kafkaProducerSettings, settings.outputTopic)
private val directory = settings.appSettings.getProperty("directory")
private val producer = Producer(settings.kafkaProducerSettings, settings.appSettings, settings.outputTopic, settings.processReportTopic)
private val recordSetId = settings.appSettings.getProperty("recordSetId")
private val sessionId = settings.appSettings.getProperty("sessionId")
private val validator = FileValidation()
private val sftpClient = SftpClient(settings.sftpSettings)
......@@ -50,34 +62,32 @@ class Service(fileName: String = "app.yml") {
sftpClient.use { sftp ->
log.info("Connected to SFTP & Kafka.")
val files = try {
val fileList = sftp.listFiles(directory).map { File(it) }
val fileList = sftp.listFiles(recordSetId).map { File(it) }
totalCount = fileList.size
fileList
} catch (ex: SftpClientException) {
ex.printStackTrace()
log.error("SFTP Exception: Could not compile the file list on sftp server in directory: $directory.")
val report = ProcessReport(
"text-file-validation",
log.error("SFTP Exception: Could not compile the file list on sftp server in directory: './$recordSetId/'.")
val report = Report(
"$recordSetId#$sessionId",
status = ReportStatus.failure,
total = 0,
successes = 0,
failures = 0
message = "SFTP Exception: ${ex.localizedMessage}"
)
producer.sendJobReport(report, settings.processReportTopic)
producer.sendReport(report)
exitProcess(1)
}
log.info("Retrieved file list from sftp server at path: $directory")
log.info("Retrieved file list from sftp server from folder: $recordSetId")
val reports = mutableListOf<Report>()
try {
log.info("There are a total of ${files.size} files to validate.")
log.info("Total files: ${files.size}.")
for (file in files) {
log.info("Validate file $file.")
log.info("Begin Validation: $file.")
val format = validator.validateExtension(file)
try {
val remoteFile = sftp.open(file)
remoteFile.use {
val validationResult = validator.validate(it.RemoteFileInputStream(), format, file)
log.info("Validated file at path ${validationResult.first}")
log.info("Validation ${validationResult.second.status}.")
producer.sendMessage(validationResult.second.id, validationResult.first)
producer.sendReport(validationResult.second)
reports.add(validationResult.second)
......@@ -95,56 +105,28 @@ class Service(fileName: String = "app.yml") {
}
}
log.info("Collected a total of ${reports.size} reports.")
val failures = reports.count { report -> report.status == ReportStatus.failure }
if (failures > 0) {
log.warn("Validation ended with $failures failures!")
producer.sendJobReport(
ProcessReport(
"text-file-validation",
status = ReportStatus.failure,
total = totalCount,
failures = failures,
successes = totalCount - failures
),
settings.processReportTopic
)
} else {
log.info("Validation was successful!")
producer.sendJobReport(
ProcessReport(
"text-file-validation",
status = ReportStatus.success,
total = totalCount,
successes = reports.size,
failures = 0
),
settings.processReportTopic
)
}
} catch (ex: SftpClientException) {
ex.printStackTrace()
log.error("SFTP Exception: ${ex.localizedMessage}.")
val report = ProcessReport(
"text-file-validation",
status = ReportStatus.failure,
total = totalCount,
failures = reports.count { report -> report.status == ReportStatus.failure },
successes = reports.count { report -> report.status == ReportStatus.success }
producer.sendReport(
Report(
"$recordSetId#$sessionId",
status = ReportStatus.failure,
message = "SFTP Exception: ${ex.localizedMessage}."
)
)
producer.sendJobReport(report, settings.processReportTopic)
exitProcess(1)
} catch (ex: Exception) {
ex.printStackTrace()
log.error(ex.javaClass.canonicalName + ": " + ex.localizedMessage)
producer.sendJobReport(
ProcessReport(
"text-file-validation",
log.error("${ex.javaClass.canonicalName}: ${ex.localizedMessage}.")
producer.sendReport(
Report(
"$recordSetId#$sessionId",
status = ReportStatus.failure,
total = totalCount,
failures = reports.count { report -> report.status == ReportStatus.failure },
successes = reports.count { report -> report.status == ReportStatus.success }
),
settings.processReportTopic
message = "Unknown Exception: ${ex.localizedMessage}."
)
)
exitProcess(1)
}
}
}
......
src/main/resources/app.yml
View file @ ebd4cfb7
......@@ -4,11 +4,20 @@ sftp:
user: ${SFTP_USER:?env}
password: ${SFTP_PASSWORD:?env}
app:
directory: ${APP_DIRECTORY:?env}
sessionId: ${SESSION_ID:?env}
recordSetId: ${RECORD_SET_ID:?env}
institutionId: ${INSTITUTION_ID:?env}
isPublished: ${IS_PUBLISHED:?env}
xmlRecordTag: ${XML_RECORD_TAG:?env}
xmlIdentifierFieldName: ${XML_IDENTIIFER_FIELD_NAME:?env}
tableSheetIndex: ${TABLE_SHEET_INDEX:?env}
tableHeaderCount: ${TABLE_HEADER_COUNT:?env}
tableHeaderIndex: ${TABLE_HEADER_INDEX:?env}
tableIdentifierIndex: ${TABLE_IDENTIFIER_INDEX:?env}
kafka:
producer:
bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:?env}
client.id: ${CLIENT_ID:?env}
topic:
out: ${TOPIC_IN:?env}
process: ${TOPIC_PROCESS:?env}
\ No newline at end of file
out: ${TOPIC_OUT:?env}
process: ${TOPIC_REPORTING:?env}
\ No newline at end of file
src/test/kotlin/TestParams.kt
View file @ ebd4cfb7
......@@ -22,6 +22,5 @@ data class TestParams(
val configFile: String,
val expectedKey: String,
val expectedValue: String,
val expectedReportValue: Report,
val expectedProcessReport: ProcessReport
val expectedReportValue: Report
)
src/test/kotlin/Tests.kt
View file @ ebd4cfb7
......@@ -17,6 +17,7 @@
*/
package org.memobase
import com.beust.klaxon.Klaxon
import java.io.FileInputStream
import java.nio.file.Paths
import java.time.Duration
......@@ -45,14 +46,14 @@ class Tests {
init {
val files = listOf(
Pair("/memobase/test_record_set_1", "brandt.csv"),
Pair("/memobase/test_record_set_2", "bauGAZH_metadaten.csv"),
Pair("/memobase/test_record_set_3", "invalid.csv"),
Pair("/memobase/test_record_set_4", "file.txt"),
Pair("/memobase/test_record_set_5", "20190906_Brandt_Metadaten.xlsx"),
Pair("/memobase/test_record_set_6", "Export_Bilder_der_Arbeit_8.csv"),
Pair("/memobase/test_record_set_7", "valid_xml.xml"),
Pair("/memobase/test_record_set_8", "invalid.xml")
Pair("/testset1", "brandt.csv"),
Pair("/testset2", "bauGAZH_metadaten.csv"),
Pair("/testset3", "invalid.csv"),
Pair("/testset4", "file.txt"),
Pair("/testset5", "20190906_Brandt_Metadaten.xlsx"),
Pair("/testset6", "Export_Bilder_der_Arbeit_8.csv"),
Pair("/testset7", "valid_xml.xml"),
Pair("/testset8", "invalid.xml")
)
for (pair in files) {
......@@ -64,6 +65,7 @@ class Tests {
}
private val consumer: KafkaConsumer<String, String>
private val klaxon = Klaxon()
init {
val props = Properties()
......@@ -74,7 +76,7 @@ class Tests {
props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.qualifiedName)
props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
consumer = KafkaConsumer(props)
consumer.subscribe(listOf("sftp-reader-p1-j1", "sftp-reader-p1-j1-reporting", "p1-reporting"))
consumer.subscribe(listOf("import-process-data-transform", "import-process-reporting"))
}
@ParameterizedTest
......@@ -84,38 +86,41 @@ class Tests {
service.run()
val topic = service.settings.outputTopic
val reportingTopic = service.settings.outputTopic + "-reporting"
val processReportingTopic = service.settings.processReportTopic
val reportingTopic = service.settings.processReportTopic
val totalConsumerRecords = mutableListOf<ConsumerRecord<String, String>>()
var result = consumer.poll(Duration.ofMillis(10))
while (totalConsumerRecords.size != 3) {
while (totalConsumerRecords.size != 2) {
if (result.count() > 0) {
totalConsumerRecords.addAll(result.asIterable())
}
log.error(result.count())
result = consumer.poll(Duration.ofMillis(10))
}
val record = totalConsumerRecords.find { value -> value.topic() == topic }!!
val recordKey = record.key()
val recordMessage = record.value()
val report = totalConsumerRecords.find { value -> value.topic() == reportingTopic }!!
val reportKey = report.key()
val reportMessage = klaxon.parse<Report>(report.value())
assertAll("",
{
assertThat(totalConsumerRecords.find { value -> value.topic() == topic })
.describedAs("Message Test")
.hasFieldOrPropertyWithValue("key", params.expectedKey)
.hasFieldOrPropertyWithValue("value", params.expectedValue)
assertThat(recordKey)
.isEqualTo(params.expectedKey)
},
{
assertThat(recordMessage)
.isEqualTo(params.expectedValue)
},
{
assertThat(totalConsumerRecords.find { value -> value.topic() == reportingTopic })
.describedAs("Report Test")
.hasFieldOrPropertyWithValue("key", params.expectedKey)
.hasFieldOrPropertyWithValue("value", params.expectedReportValue.toJson())
assertThat(reportKey)
.isEqualTo(params.expectedKey)
},
{
assertThat(totalConsumerRecords.find { value -> value.topic() == processReportingTopic })
.describedAs("Process Report Test")
.hasFieldOrPropertyWithValue("key", params.expectedProcessReport.id)
.hasFieldOrPropertyWithValue("value", params.expectedProcessReport.toJson())
assertThat(reportMessage)
.isEqualTo(params.expectedReportValue)
}
)
}
......@@ -124,137 +129,81 @@ class Tests {
TestParams(
"test1.yml",
expectedKey = "brandt.csv",
expectedValue = "{\"format\" : \"CSV\", \"path\" : \"/memobase/test_record_set_1/brandt.csv\"}",
expectedValue = "{\"format\" : \"CSV\", \"path\" : \"/testset1/brandt.csv\"}",
expectedReportValue = Report(
id = "brandt.csv",
status = "SUCCESS",
message = "Validated file at path /memobase/test_record_set_1/brandt.csv with format CSV."
),
expectedProcessReport = ProcessReport(
id = "text-file-validation",
status = ReportStatus.success,
total = 1,
failures = 0,
successes = 1
message = "Validated file at path /testset1/brandt.csv with format CSV."
)
),
TestParams(
"test2.yml",
expectedKey = "bauGAZH_metadaten.csv",
expectedValue = "{\"format\" : \"CSV\", \"path\" : \"/memobase/test_record_set_2/bauGAZH_metadaten.csv\"}",
expectedValue = "{\"format\" : \"CSV\", \"path\" : \"/testset2/bauGAZH_metadaten.csv\"}",
expectedReportValue = Report(
id = "bauGAZH_metadaten.csv",
status = "SUCCESS",
message = "Validated file at path /memobase/test_record_set_2/bauGAZH_metadaten.csv with format CSV."
),
expectedProcessReport = ProcessReport(
id = "text-file-validation",
status = "SUCCESS",
total = 1,
failures = 0,
successes = 1
message = "Validated file at path /testset2/bauGAZH_metadaten.csv with format CSV."
)
),
TestParams(
"test3.yml",
expectedKey = "invalid.csv",
expectedValue = "{\"format\" : \"ERROR\", \"path\" : \"/memobase/test_record_set_3/invalid.csv\"}",
expectedValue = "{\"format\" : \"ERROR\", \"path\" : \"/testset3/invalid.csv\"}",
expectedReportValue = Report(
id = "invalid.csv",
status = "FAILURE",
message = "CSV ERROR: Fields num seems to be 4 on each row, but on 2th csv row, fields num is 6. for file /memobase/test_record_set_3/invalid.csv."
),
expectedProcessReport = ProcessReport(
id = "text-file-validation",
status = "FAILURE",
total = 1,
failures = 1,
successes = 0
message = "CSV ERROR: Fields num seems to be 4 on each row, but on 2th csv row, fields num is 6. for file /testset3/invalid.csv."
)
),
TestParams(
"test4.yml",
expectedKey = "file.txt",
expectedValue = "{\"format\" : \"ERROR\", \"path\" : \"/memobase/test_record_set_4/file.txt\"}",
expectedValue = "{\"format\" : \"ERROR\", \"path\" : \"/testset4/file.txt\"}",
expectedReportValue = Report(
id = "file.txt",
status = "FAILURE",
message = "File Extension Error: Not a valid file extension: file.txt."
),
expectedProcessReport = ProcessReport(
id = "text-file-validation",
status = "FAILURE",
total = 1,
failures = 1,
successes = 0
)
),
TestParams(
"test5.yml",
expectedKey = "20190906_Brandt_Metadaten.xlsx",
expectedValue = "{\"format\" : \"XLSX\", \"path\" : \"/memobase/test_record_set_5/20190906_Brandt_Metadaten.xlsx\"}",
expectedValue = "{\"format\" : \"XLSX\", \"path\" : \"/testset5/20190906_Brandt_Metadaten.xlsx\"}",
expectedReportValue = Report(
id = "20190906_Brandt_Metadaten.xlsx",
status = "SUCCESS",
message = "Validated file at path /memobase/test_record_set_5/20190906_Brandt_Metadaten.xlsx with format XLSX."
),
expectedProcessReport = ProcessReport(
id = "text-file-validation",
status = "SUCCESS",
total = 1,
failures = 0,
successes = 1
message = "Validated file at path /testset5/20190906_Brandt_Metadaten.xlsx with format XLSX."
)
),
TestParams(
"test6.yml",