✨ :feat: Replace tofhir-log-server with Fluentd, Elasticsearch and Ki…

…bana stack

README.md

@@ -14,7 +14,6 @@ be either persisted to a file system or to a running HL7 FHIR endpoint.
 toFHIR consists of the following modules:
 - `tofhir-engine`: The core module of toFHIR which includes the main functionality of the tool.
 - `tofhir-server`: A standalone web server module that provides a REST API to run mapping jobs via `tohir-engine` and manage the mapping job definitions.
-- `tofhir-log-server`: A standalone web server module that provides a REST API to query the logs of the mapping job execution results.
 - `tofhir-server-common`: Holds common files like server configuration or errors for server implementations.
 - `tofhir-common`: Contains model and utility classes shared across various modules.
 - `tofhir-rxnorm`: Provides a client implementation to access the RxNorm API and a FHIR Path Function library to utilize its API functionalities in mapping definitions.
@@ -272,11 +271,6 @@ webserver = {
     password = null
   }
 }
-
-# The service from where tofhir-server will read the logs.
-log-service = {
-  endpoint = "http://localhost:8086/tofhir-logs"
-}
 ```
 
 After the server is up and running, the engine will be available via the REST API.
@@ -1221,3 +1215,32 @@ Or both. This will delete the source files after processing/mapping and save the
 While `archiveMode` works on a file-based basis, `saveErroneousRecords` works for each record/row in the source data.
 
 Please also note that, the `archiveMode` config is only applicable for the file system source type.
+
+## Mapping Job Results with EFK Stack
+This project utilizes the EFK Stack (Elasticsearch, Fluentd, and Kibana) to visualize the results of mapping job executions. 
+Once the EFK Stack is started using the provided [docker-compose.yml](docker/docker-compose.yml), Kibana can be accessed at http://localhost:5601.
+### Initializing Kibana
+Follow these steps to initialize Kibana with predefined configurations, dashboards, and indexes:
+#### Importing Visualizations, Dashboards, and Indexes
+1. Navigate to http://localhost:5601/app/management/kibana/objects
+2. Click on the `Import` button and import [export.ndjson](docker/kibana-data/export.ndjson)
+
+![Importing Objects](readme-assets%2Fkibana-saved-objects.png)
+#### Creating an Index Template for 'ignore_above' Properties of Strings
+To handle potentially long log messages, increase the default value of `ignore_above` for string properties.
+
+1. Go to http://localhost:5601/app/dev_tools#/console
+2. Copy and paste the contents of [index_template.txt](docker/kibana-data/index_template.txt) into the console and execute the request.
+
+![Index Template API Response](readme-assets%2Fkibana-index-template.png)
+
+### Kibana Dashboards
+After running mapping jobs, you can view their logs via Kibana dashboards.
+### Executions Dashboard
+This dashboard provides an overview of each execution's result.
+![Executions Dashboard](readme-assets%2Fkibana-executions-dashboard.png)
+To analyze a specific execution, hover over the execution ID and click on the plus icon next to it. Then, select the `Go to Dashboard` option as shown below:
+![Go to Dashboard](readme-assets%2Fkibana-go-to-dashboard.png)
+### Execution Details Dashboard
+This dashboard displays the results of individual mapping tasks and any corresponding errors.
+![Execution Details Dashboard](readme-assets%2Fkibana-execution-details.png)

docker/docker-compose.yml

@@ -1,21 +1,10 @@
 version: '3.9'
 services:
-  tofhir-log-server:
-    image: srdc/tofhir-log-server:latest
-    container_name: tofhir-log-server
-    hostname: tofhir-log-server
-    ports:
-      - "8086:8086"
-    networks:
-      - tofhir-network
-    volumes:
-      - './tofhir-docker-logs:/usr/local/tofhir/logs'
   tofhir-server:
     image: srdc/tofhir-server:latest
     container_name: tofhir-server
     hostname: tofhir-server
     environment:
-      - LOG_SERVICE_ENDPOINT=http://tofhir-log-server:8086/tofhir-logs
       - FHIR_DEFINITIONS_ENDPOINT=http://onfhir:8080/fhir
       - FHIR_REPO_URL=http://onfhir:8080/fhir
       - APP_CONF_FILE=/usr/local/tofhir/conf/docker/tofhir-server.conf
@@ -34,7 +23,44 @@ services:
       - "8087:80"
     networks:
       - tofhir-network
+ # Elasticsearch, Fluentd and Kibana stack for mapping job log management
+  elasticsearch:
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.15.2
+    container_name: elasticsearch
+    environment:
+      - discovery.type=single-node
+    ports:
+      - "9200:9200"
+    networks:
+      - tofhir-network
+    volumes:
+      - elasticsearch_data:/usr/share/elasticsearch/data
+  fluentd:
+    build: ./fluentd
+    container_name: fluentd
+    volumes:
+      - ./fluentd/conf:/fluentd/etc
+    ports:
+      - "24224:24224"
+      - "24224:24224/udp"
+    networks:
+      - tofhir-network
+  # TODO: Investigate if it's possible to populate predefined Kibana visualizations, dashboards, and indexes
+  # using Docker. This could involve making API calls or utilizing volumes.
+  # The current approach is the utilize Kibana UI as described in the README file.
+  kibana:
+    image: docker.elastic.co/kibana/kibana:7.15.2
+    container_name: kibana
+    ports:
+      - "5601:5601"
+    networks:
+      - tofhir-network
+    depends_on:
+      - elasticsearch
 networks:
   tofhir-network:
     name: onfhir-network
     external: true
+volumes:
+  elasticsearch_data:
+    driver: local

docker/fluentd/Dockerfile

@@ -0,0 +1,15 @@
+#Specifies the base image for building the Fluentd Docker image
+FROM fluent/fluentd:v1.11-1
+#Switches to the root user account to perform package installations and configurations.
+USER root
+#Installs necessary packages and dependencies using the Alpine package manager (apk).
+RUN apk add --no-cache --update --virtual .build-deps \
+    sudo build-base ruby-dev \
+    && sudo gem install faraday -v 2.8.1 \
+    && sudo gem install faraday-net_http -v 3.0.2 \
+    && sudo gem install fluent-plugin-elasticsearch \
+    && sudo gem sources --clear-all \
+    && apk del .build-deps \
+    && rm -rf /tmp/* /var/tmp/* /usr/lib/ruby/gems/*/cache/*.gem
+# Switches back to the fluent user account to run Fluentd.
+USER fluent

docker/fluentd/conf/fluent.conf

@@ -0,0 +1,23 @@
+// Defines a Fluentd configuration for forwarding logs to Elasticsearch.
+
+<source>
+  // Specifies the input plugin type as 'forward', which receives log entries forwarded by Fluentd clients.
+  @type forward
+  // Specifies the port on which the forward input plugin listens for incoming log entries.
+  port 24224
+</source>
+
+<match **>
+  // Specifies the output plugin type as 'elasticsearch', which sends log entries to an Elasticsearch cluster.
+  @type elasticsearch
+  // Specifies the host address of the Elasticsearch cluster.
+  host elasticsearch
+  // Specifies the port number of the Elasticsearch cluster.
+  port 9200
+  // Configures the output format to be compatible with Logstash.
+  logstash_format true
+  // Specifies the prefix to be added to the index name in Elasticsearch.
+  logstash_prefix fluentd
+  // Specifies the interval at which buffered log entries are flushed to Elasticsearch.
+  flush_interval 5s
+</match>

docker/kibana-data/index_template.txt

@@ -0,0 +1,24 @@
+PUT _template/ignore_above
+{
+  "index_patterns": [
+    "fluentd*"
+  ],
+  "mappings": {
+    "dynamic_templates": [
+      {
+        "strings_as_keyword": {
+          "match_mapping_type": "string",
+          "mapping": {
+            "type": "text",
+            "fields": {
+              "keyword": {
+                "ignore_above": 10000,
+                "type": "keyword"
+              }
+            }
+          }
+        }
+      }
+    ]
+  }
+}

docker/server/docker-entrypoint.sh

@@ -74,11 +74,6 @@ if [ ! -z "$WEBSERVER_BASEURI" ]; then
   JAVA_CMD+="-Dwebserver.base-uri=$WEBSERVER_BASEURI "
 fi
 
-# Configure log service
-if [ ! -z "$LOG_SERVICE_ENDPOINT" ]; then
-  JAVA_CMD+="-Dlog-service.endpoint=$LOG_SERVICE_ENDPOINT "
-fi
-
 # Delay the execution for this amount of seconds
 if [ ! -z "$DELAY_EXECUTION" ]; then
   sleep $DELAY_EXECUTION

pom.xml

@@ -78,7 +78,6 @@
     <modules>
         <module>tofhir-engine</module>
         <module>tofhir-server</module>
-        <module>tofhir-log-server</module>
         <module>tofhir-common</module>
         <module>tofhir-server-common</module>
         <module>tofhir-rxnorm</module>
@@ -101,6 +100,8 @@
         <onfhir-template-engine.version>1.1-SNAPSHOT</onfhir-template-engine.version>
         <spark-on-fhir.version>1.0-SNAPSHOT</spark-on-fhir.version>
         <json4s.version>3.7.0-M11</json4s.version>
+        <logback-more-appenders.version>1.8.8</logback-more-appenders.version>
+        <fluent-logger.version>0.3.4</fluent-logger.version>
         <com.fasterxml.version>2.15.1</com.fasterxml.version>
         <scalatest.version>3.2.17</scalatest.version>
         <spark.version>3.5.1</spark.version>
@@ -280,6 +281,22 @@
 <!--                    </exclusion>-->
 <!--                </exclusions>-->
             </dependency>
+            <!-- Dependency for Fluentd-Logback Integration -->
+            <!-- This library provides additional appenders (namely, ch.qos.logback.more.appenders.DataFluentAppender)
+             for Logback, extending its capabilities.-->
+            <dependency>
+                <groupId>com.sndyuk</groupId>
+                <artifactId>logback-more-appenders</artifactId>
+                <version>${logback-more-appenders.version}</version>
+            </dependency>
+            <!-- This dependency provides access to the Fluentd Logger Library, which allows the applications to easily
+             send logs to Fluentd-->
+            <dependency>
+                <groupId>org.fluentd</groupId>
+                <artifactId>fluent-logger</artifactId>
+                <version>${fluent-logger.version}</version>
+            </dependency>
+            <!-- The end of dependencies for Fluentd-Logback Integration-->
 
             <!-- These are required for Spark and resolving among different versions coming from akka and json4s -->
             <!-- Dependency from jackson-dataformat-csv and logstash-logback-encoder-->
@@ -418,11 +435,6 @@
                 <artifactId>tofhir-server-common_2.13</artifactId>
                 <version>${project.version}</version>
             </dependency>
-            <dependency>
-                <groupId>io.onfhir</groupId>
-                <artifactId>tofhir-log-server-common_2.13</artifactId>
-                <version>${project.version}</version>
-            </dependency>
             <dependency>
                 <groupId>io.onfhir</groupId>
                 <artifactId>tofhir-engine_2.13</artifactId>

tofhir-engine/pom.xml

@@ -200,6 +200,15 @@
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-sql_${scala.binary.version}</artifactId>
         </dependency>
+        <!-- Fluentd-Logback Integration-->
+        <dependency>
+            <groupId>com.sndyuk</groupId>
+            <artifactId>logback-more-appenders</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.fluentd</groupId>
+            <artifactId>fluent-logger</artifactId>
+        </dependency>
 
         <!-- Application Configuration -->
         <dependency>

tofhir-engine/src/main/resources/logback.xml

@@ -45,6 +45,9 @@
                 <levelValue>[ignore]</levelValue>
             </fieldNames>
         </encoder>
+        <!-- Encoder to map MapMarker to LogstashMarker
+             Without this encoder, RollingFileAppender will ignore the markers -->
+        <encoder class="io.tofhir.engine.logback.MapMarkerToLogstashMarkerEncoder" />
 
         <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
             <FileNamePattern>logs/tofhir-mappings.%i.log.zip</FileNamePattern>
@@ -57,6 +60,19 @@
         </triggeringPolicy>
     </appender>
 
+    <!-- Appender for Fluentd -->
+    <appender name="FLUENT" class="ch.qos.logback.more.appenders.DataFluentAppender">
+        <!-- Configure the host and port for fluentd -->
+        <remoteHost>localhost</remoteHost>
+        <port>24224</port>
+        <!-- We need only INFO logs. -->
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>INFO</level>
+        </filter>
+        <!-- If true, Map Marker is expanded instead of nesting in the marker name -->
+        <flattenMapMarker>true</flattenMapMarker>
+    </appender>
+
     <appender name="ASYNC-AUDIT" class="ch.qos.logback.classic.AsyncAppender">
         <appender-ref ref="FILE-AUDIT" />
     </appender>
@@ -69,6 +85,7 @@
 
     <logger name="io.tofhir.engine.data.write.SinkHandler" level="INFO">
         <appender-ref ref="ASYNC-AUDIT" />
+        <appender-ref ref="FLUENT" />
     </logger>
 
     <!-- Give me DEBUG level logs from io.tofhir package because the default is set to ERROR at root (above) -->

tofhir-engine/src/main/scala/io/tofhir/engine/data/write/SinkHandler.scala

@@ -90,18 +90,18 @@ object SinkHandler {
 
     //Log the job result
     val jobResult = FhirMappingJobResult(mappingJobExecution, mappingUrl, numOfInvalids, numOfNotMapped, numOfWritten, numOfNotWritten)
-    logger.info(jobResult.toLogstashMarker, jobResult.toString)
+    logger.info(jobResult.toMapMarker, jobResult.toString)
 
     // Log the mapping and invalid input errors
     if (numOfNotMapped > 0 || numOfInvalids > 0) {
       mappingErrors.union(invalidInputs).foreach(r =>
-        logger.warn(r.copy(executionId = Some(mappingJobExecution.id)).toLogstashMarker,
+        logger.warn(r.copy(executionId = Some(mappingJobExecution.id)).toMapMarker,
           r.copy(executionId = Some(mappingJobExecution.id)).toString)
       )
     }
     if (numOfNotWritten > 0)
       notWrittenResources.forEach(r =>
-        logger.warn(r.copy(executionId = Some(mappingJobExecution.id)).toLogstashMarker,
+        logger.warn(r.copy(executionId = Some(mappingJobExecution.id)).toMapMarker,
           r.copy(executionId = Some(mappingJobExecution.id)).toString)
       )
   }