add logstash files

Author: c-neto

.gitignore

@@ -1,2 +1,3 @@
 venv/
 .idea
+.DS_Store

logstash/db-enrichment/Makefile

@@ -3,6 +3,3 @@ up:
 
 down:
 	docker-compose down
-
-test:
-	./test.sh

logstash/db-enrichment/README.md

@@ -0,0 +1,101 @@
+# DB Enrichment Project
+
+This project sets up a local environment for enriching logs with data from a PostgreSQL database using Logstash. The setup includes Docker containers for Logstash and PostgreSQL, along with the necessary configuration files and scripts. The purpose of this project is to test JDBC connection string parameters to avoid Logstash's bad behavior when the PostgreSQL table is locked or slow.
+
+The following sections provide detailed instructions on how to set up and use the provided Docker containers, as well as how to send logs for enrichment.
+
+```bash
+├── Makefile                    # Docker-compose shortcut (up|down)
+├── docker-compose.yaml         # Docker-compose manifest file
+├── logstash
+│   ├── Dockerfile              # Logstash Dockerfile
+│   ├── config
+│   │   ├── logstash.yml        # Logstash configuration file
+│   │   └── pipelines.yml       # Logstash pipelines configuration
+│   └── pipeline
+│       └── pipeline.cfg        # Main Logstash pipeline configuration
+├── postgres
+│   └── init.sql                # PostgreSQL initialization script
+└── send-log.sh                 # Script to send logs to Logstash
+```
+
+# Set up the containers
+
+To start the containers, use the following command:
+
+```bash
+make up
+```
+
+To stop the containers, use the following command:
+
+```bash
+make down
+```
+
+# Send logs
+
+The script [send-log.sh](./send-log.sh) is used to send logs to Logstash through HTTP. There are two types of logs: enriched logs and non-enriched logs.
+
+To send logs that will be enriched from the database, use the following command:
+
+```shell
+./send-log.sh db
+```
+
+To send logs that will not be enriched, use the following command:
+
+```shell
+./send-log.sh
+```
+
+# JDBC String Connection Details
+
+More details about JDBC_CONNECTION_STRING parameters:
+
+- https://jdbc.postgresql.org/documentation/use/
+- https://www.postgresql.org/docs/current/runtime-config-client.html
+
+JDBC connection string used in this stack:
+
+```
+jdbc:postgresql://postgres:5432/db?ApplicationName=logstash&loginTimeout=10&socketTimeout=10&options=-c%20statement_timeout=5000%20-c%20lock_timeout=1000
+```
+
+- `ApplicationName=logstash`: Sets the application name to "logstash" for easier identification in PostgreSQL logs.
+- `loginTimeout=10`: Specifies the maximum time in seconds to wait for a connection to be established. If the connection cannot be established within this time, an error is thrown.
+- `socketTimeout=10`: Sets the maximum time in seconds for reading data from the database. If the database does not respond within this time, the connection is closed.
+- `options=-c statement_timeout=5000 -c lock_timeout=1000`: Passes additional configuration parameters to PostgreSQL:
+    - `statement_timeout=5000`: Sets the maximum time in milliseconds that a query can run before being terminated. In this case, queries running longer than 5 seconds will be aborted.
+    - `lock_timeout=1000`: Sets the maximum time in milliseconds to wait for a lock on a table. If the lock cannot be acquired within 1 second, an error is thrown.
+
+# SQL Command to Lock Table
+
+With auto-commit off, when you execute the command below, the table will be locked. When you commit, the table will be unlocked.
+
+```sql
+LOCK TABLE table_users IN ACCESS EXCLUSIVE MODE;
+```
+
+# How to Simulate a Slow Query
+
+To test how Logstash handles slow queries, you can simulate a delay in the PostgreSQL response using the `pg_sleep` function. This function pauses the execution for a specified number of seconds. In the example below, the query will sleep for 7 seconds before returning the results.
+
+Use the following SQL command to simulate a slow query:
+
+```sql
+SELECT
+    pg_sleep(7)::text,
+    user_name,
+    user_email,
+    user_group
+FROM table_users
+WHERE id = ?
+```
+
+In this query:
+- `pg_sleep(7)::text` introduces a 7-second delay.
+- `user_name`, `user_email`, and `user_group` are the columns being retrieved from the `table_users` table.
+- `id = ?` is a placeholder for the user ID you want to query.
+
+This setup helps you observe how Logstash behaves when the database query takes longer to execute.

logstash/db-enrichment/docker-compose.yaml

@@ -13,28 +13,21 @@ services:
       - ./postgres/init.sql:/docker-entrypoint-initdb.d/init.sql
 
   logstash:
-    image: docker.elastic.co/logstash/logstash:8.11.2
     container_name: logstash
     restart: always
     build:
       context: logstash
+      args:
+        LOGSTASH_IMAGE: docker.elastic.co/logstash/logstash-oss:8.9.2
+        JDBC_JAR_DOWNLOAD_URL: https://jdbc.postgresql.org/download/postgresql-42.5.0.jar
     environment:
       DB_USER: user
       DB_PASSWORD: password
-      DB_JDBC_CONNECTION_STRING: jdbc:postgresql://postgres:5432/users_logs
-      # DB_JDBC_CONNECTION_STRING: jdbc:postgresql://postgres:5432/users_logs?ApplicationName=logstash&loginTimeout=10&socketTimeout=10&options=-c%20statement_timeout=5000%20-c%20lock_timeout=1000
+      # DB_JDBC_CONNECTION_STRING: jdbc:postgresql://postgres:5432/db
+      DB_JDBC_CONNECTION_STRING: jdbc:postgresql://postgres:5432/db?ApplicationName=logstash&loginTimeout=10&socketTimeout=10&options=-c%20statement_timeout=5000%20-c%20lock_timeout=1000
     ports:
       - 8090:8090
     volumes:
       - ./logstash/pipeline:/usr/share/logstash/pipeline
       - ./logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml
       - ./logstash/config/pipelines.yml:/usr/share/logstash/config/pipelines.yml
-
-  adminer:
-    image: adminer
-    container_name: adminer
-    restart: always
-    ports:
-      - "8080:8080"
-    environment:
-      ADMINER_DEFAULT_SERVER: postgres

logstash/db-enrichment/logstash/Dockerfile

@@ -1,5 +1,10 @@
-FROM docker.elastic.co/logstash/logstash:8.9.2
+# Define build arguments the specified Logstash image
+ARG LOGSTASH_IMAGE
+FROM ${LOGSTASH_IMAGE}
 
+# Set the working directory to /usr/share/logstash
 WORKDIR /usr/share/logstash
 
-RUN curl -o /usr/share/logstash/postgresql-42.5.0.jar https://jdbc.postgresql.org/download/postgresql-42.5.0.jar
+# Download the PostgreSQL JDBC driver and save it as postgresql.jar in the working directory
+ARG JDBC_JAR_DOWNLOAD_URL
+RUN curl -L "${JDBC_JAR_DOWNLOAD_URL}" -o /usr/share/logstash/postgresql.jar

logstash/db-enrichment/logstash/pipeline/pipeline.cfg

@@ -1,3 +1,6 @@
+# SQL command to lock table
+# lock table table_users IN ACCESS exclusive mode;
+
 input {
   http {
     port => 8090
@@ -11,39 +14,46 @@ input {
 
 filter {
   mutate {
-    remove_field => ["headers"]
+    remove_field => ["headers", "original", "event", "host"]
   }
 
-  jdbc_streaming {
-    jdbc_connection_string => "${DB_JDBC_CONNECTION_STRING}"
-    jdbc_user => "${DB_USER}"
-    jdbc_password => "${DB_PASSWORD}"
-    jdbc_validate_connection => true
-    jdbc_validation_timeout => 1
-    jdbc_driver_library => "/usr/share/logstash/postgresql-42.5.0.jar"
-    jdbc_driver_class => "org.postgresql.Driver"
-    # https://github.com/jeremyevans/sequel/blob/master/doc/opening_databases.rdoc#top
-    # sequel_opts => {
-    #   max_connections => 4
-    #   pool_timeout => 5
-    # }
-    # pg_sleep(10)::text,
-    statement => "
-      SELECT
-        user_name,
-        user_email,
-        user_group
-      FROM db_users
-      WHERE id = :input_parameter
-    "
-    parameters => {
-      "input_parameter" => "[document][user_id]"
+  if [document][user_id] {
+
+    jdbc_streaming {
+      jdbc_driver_library => "/usr/share/logstash/postgresql.jar"
+      jdbc_driver_class => "org.postgresql.Driver"
+      jdbc_connection_string => "${DB_JDBC_CONNECTION_STRING}"
+      jdbc_user => "${DB_USER}"
+      jdbc_password => "${DB_PASSWORD}"
+      use_prepared_statements => true
+      prepared_statement_name => "logstash_enrich_query"
+      prepared_statement_bind_values => ["[document][user_id]"]
+      statement => "
+        SELECT
+          user_name,
+          user_email,
+          user_group
+        FROM table_users
+        WHERE id = ?
+      "
+      target => "sql"
     }
-    target => "sql"
+
   }
 
 }
 
 output {
   stdout { codec => rubydebug }
+  # stdout { codec => json }
 }
+
+# jdbc_validate_connection => true
+# jdbc_validation_timeout => 1
+# https://github.com/jeremyevans/sequel/blob/master/doc/opening_databases.rdoc#top
+# sequel_opts => {
+#   max_connections => 4
+#   pool_timeout => 5
+# }
+# pg_sleep(10)::text,
+

logstash/db-enrichment/postgres/init.sql

@@ -1,15 +1,15 @@
-CREATE DATABASE users_logs;
+CREATE DATABASE db;
 
-\c users_logs;
+\c db;
 
-CREATE TABLE db_users (
+CREATE TABLE table_users (
     id SERIAL PRIMARY KEY,
     user_name VARCHAR(100) NOT NULL,
     user_email VARCHAR(100) NOT NULL UNIQUE,
     user_group VARCHAR(50) NOT NULL
 );
 
-INSERT INTO db_users (user_name, user_email, user_group) VALUES
+INSERT INTO table_users (user_name, user_email, user_group) VALUES
 ('Alice Silva', '[email protected]', 'admin'),
 ('Bruno Costa', '[email protected]', 'user'),
 ('Carlos Mendes', '[email protected]', 'user'),

logstash/db-enrichment/send-log.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+
+# Set the URL and headers
+URL="http://127.0.0.1:8090/"
+HEADER="Content-Type: application/json"
+
+# Function to send a POST request
+send_post_request() {
+  local payload=$1
+  curl -X POST "$URL" \
+    -H "$HEADER" \
+    -d "$payload"
+}
+
+# Send the appropriate payload based on the argument
+if [[ "$1" == "db" ]]; then
+  send_post_request '{"event": "purchase", "user_id": 3}'
+else
+  send_post_request '{"event": "no database query"}'
+fi