Threat Search & Observability Platform

Add full-text search, metrics, and structured logging to a threat pipeline.

Introduction

30 Second Summary

When a security alert fires, analysts need to search through thousands of events to find related activity. Scrolling through database tables one query at a time is painfully slow when you need answers in seconds.

In this project, you will build a search and observability layer on top of your existing Threat Event Pipeline. You will add Elasticsearch for full-text search, Kibana for visual exploration, Prometheus and Grafana for real-time metrics, and structured JSON logs for machine-parseable output.

What You'll Build

You will type a search term like "brute force" into your API and instantly see matching threat events pulled from across descriptions, usernames, and source IPs. Then you will open Kibana to visually drill into patterns and watch live Grafana charts track every query you run.

By the end of this project, you'll have:

  • A full-text search API where you can search by keyword across multiple fields and combine it with filters for severity, source IP, event type, and date range.
  • A live Kibana investigation workflow where you visually explore threat patterns, drill into individual events, and build saved searches.
  • A Grafana monitoring dashboard showing search latency percentiles, request throughput, error rates, and JVM memory usage updating in real time.
  • Automated integration tests that spin up a real Elasticsearch container and verify the entire index-and-search flow without mocking.
  • Secret Mission: Add distributed tracing with Zipkin to correlate requests across the entire pipeline and visualize span timings.

Are there any prerequisites?

This project builds directly on Project 1 (Threat Event Pipeline). You need a working Spring Boot application with ThreatEvent and ThreatAlert entities, PostgreSQL, Kafka, and Docker Compose already configured.

You also need Docker Desktop with at least 4GB of memory allocated, Java 17+, and Maven installed on your machine.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters.

Your existing threat pipeline ingests events and generates alerts, but it has no way to search across fields or tell you how it's performing. In this project, you'll add a full search and observability layer on top of that pipeline using Elasticsearch, Kibana, Prometheus, and Grafana.

Set Up the Observability Stack

Your threat pipeline stores events in PostgreSQL and streams them through Kafka. That handles writes and processing well, but offers no way to search across fields or monitor how the system performs under load.

In this step, you will add the infrastructure that makes your pipeline searchable and observable. Elasticsearch gives you sub-second full-text search. Kibana lets you explore data visually. Prometheus and Grafana provide real-time metrics dashboards.

In this step, get ready to:
  • Add Elasticsearch, Kibana, Prometheus, and Grafana to your Docker Compose stack.
  • Add Maven dependencies for search, metrics, structured logging, and integration testing.
  • Configure Spring Boot to connect to Elasticsearch and expose Prometheus metrics.
Add search and monitoring services to Docker Compose

Your existing docker-compose.yml runs PostgreSQL, Kafka, and Zookeeper. You will add four new services to it.

  • Open docker-compose.yml in your IDE and add the Elasticsearch service below your existing services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.18.0
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - xpack.security.enrollment.enabled=false
      - ES_JAVA_OPTS=-Xms512m -Xmx512m
      - bootstrap.memory_lock=true
    ulimits:
      memlock:
        soft: -1
        hard: -1
    ports:
      - "9200:9200"
    volumes:
      - esdata:/usr/share/elasticsearch/data
    networks:
      - threat-network
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:9200/_cluster/health || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 10

What does this configuration do?

The discovery.type=single-node setting tells Elasticsearch it does not need to discover other cluster nodes. This avoids bootstrap checks that would fail in a single-container setup.

Security is disabled with xpack.security.enabled=false because configuring TLS certificates adds complexity with no learning value for this project. The heap is capped at 512MB to keep total Docker memory reasonable.

This security configuration is for local development only

In production, you must enable TLS, set strong passwords, and use Elasticsearch's built-in role-based access control. Never disable security on a cluster that handles real data.

  • Save docker-compose.yml and start Elasticsearch by running:
docker compose up -d elasticsearch

Elasticsearch takes 20-40 seconds to become healthy. Wait a moment, then verify it is responding.

  • Check the cluster health by running:
curl http://localhost:9200/_cluster/health?pretty

You should see a JSON response with "status" : "green" and "number_of_nodes" : 1. This confirms Elasticsearch is running as a healthy single-node cluster.

Elasticsearch container exits with code 137?

Exit code 137 means Docker killed the container for exceeding its memory limit. Open Docker Desktop, go to Settings then Resources, and increase memory to at least 6GB.

Alternatively, reduce the ES heap to -Xms256m -Xmx256m in the environment variables, or stop your Project 1 Kafka containers to free RAM.

Help me fix my Elasticsearch memory issue

  • Now add the Kibana service directly below the elasticsearch service in docker-compose.yml:
  kibana:
    image: docker.elastic.co/kibana/kibana:8.18.0
    container_name: kibana
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    ports:
      - "5601:5601"
    depends_on:
      elasticsearch:
        condition: service_healthy
    networks:
      - threat-network

What does this configuration do?

Kibana connects to Elasticsearch using the Docker network hostname elasticsearch (the container name). The depends_on with condition: service_healthy ensures Kibana only starts after the ES healthcheck passes.

  • Save docker-compose.yml and bring up Kibana by running:
docker compose up -d
  • Open http://localhost:5601 in your browser.

You should see the Kibana home page. It may take 30-60 seconds to become ready on first launch.

  • Next, create a new file called prometheus.yml in your project root (the same directory as docker-compose.yml). Add the following scrape configuration:
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: "threat-pipeline"
    metrics_path: "/actuator/prometheus"
    static_configs:
      - targets: ["host.docker.internal:8080"]

What does this configuration do?

Prometheus will poll your Spring Boot app every 15 seconds at the /actuator/prometheus endpoint. The target host.docker.internal:8080 is a special Docker Desktop hostname that resolves to the host machine, since your Spring Boot app runs on the host while Prometheus runs inside a container.

  • Now add the Prometheus and Grafana services plus the volume declarations to docker-compose.yml, below the kibana service:
  prometheus:
    image: prom/prometheus:v3.13.0
    container_name: prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    ports:
      - "9090:9090"
    networks:
      - threat-network

  grafana:
    image: grafana/grafana:13.0.3
    container_name: grafana
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana_storage:/var/lib/grafana
    ports:
      - "3000:3000"
    depends_on:
      - prometheus
    networks:
      - threat-network

volumes:
  esdata:
  prometheus_data:
  grafana_storage:

What does this configuration do?

Prometheus mounts the prometheus.yml you just created so it knows what to scrape. Grafana starts with a default admin password of admin and persists dashboards in a named volume.

The three new volumes (esdata, prometheus_data, grafana_storage) keep data intact across container restarts.

  • Save docker-compose.yml and bring up all services by running:
docker compose up -d
  • Confirm all containers are running by running:
docker compose ps

You should see elasticsearch, kibana, prometheus, and grafana all showing a running state alongside your existing services.

Prometheus target shows DOWN?

This is expected right now because your Spring Boot app is not running yet. Prometheus will show the target as DOWN until you start the application in substep 3. The container itself should be running.

If the prometheus container is not running at all, check that prometheus.yml exists in the same directory as docker-compose.yml. A missing file causes a mount error.

Help me fix my Prometheus container

✔️ Awesome, I've got everything!

Great. Make sure you have saved both docker-compose.yml and prometheus.yml.

ⓧ I'd like to double check the full code

Your docker-compose.yml additions are spread across multiple chunks above. Scroll up and confirm these four service blocks are present in order: elasticsearch, kibana, prometheus, grafana. Also confirm the volumes block at the bottom declares esdata, prometheus_data, and grafana_storage.

Your prometheus.yml should contain:

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: "threat-pipeline"
    metrics_path: "/actuator/prometheus"
    static_configs:
      - targets: ["host.docker.internal:8080"]
Add Maven dependencies

Your project needs libraries to talk to Elasticsearch, expose Prometheus metrics, produce structured logs, and run integration tests with real containers.

  • Open pom.xml and add these dependencies inside the <dependencies> block:
<!-- Elasticsearch -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

<!-- Prometheus metrics -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

<!-- Structured JSON logging -->
<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>8.1</version>
</dependency>

<!-- Test: Elasticsearch container -->
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>testcontainers-elasticsearch</artifactId>
    <version>2.0.5</version>
    <scope>test</scope>
</dependency>

What do these dependencies do?

  • spring-boot-starter-data-elasticsearch provides Spring Data repositories and the Elasticsearch Java client for indexing and querying documents.
  • micrometer-registry-prometheus auto-registers a Prometheus metrics endpoint at /actuator/prometheus.
  • logstash-logback-encoder version 8.1 converts log output to structured JSON. Version 9.0 requires Jackson 3.x, which is incompatible with Spring Boot 3.5.x.
  • testcontainers-elasticsearch spins up a real Elasticsearch container for integration tests without needing your Docker Compose stack running.
  • Save pom.xml.

✔️ Awesome, I've got everything!

Great. Make sure pom.xml is saved. Your IDE may start downloading the new dependencies automatically.

ⓧ I'd like to double check the full code

Confirm these four dependency blocks are present inside your <dependencies> section:

<!-- Elasticsearch -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

<!-- Prometheus metrics -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

<!-- Structured JSON logging -->
<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>8.1</version>
</dependency>

<!-- Test: Elasticsearch container -->
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>testcontainers-elasticsearch</artifactId>
    <version>2.0.5</version>
    <scope>test</scope>
</dependency>
Configure Spring Boot and verify the full stack

Spring Boot needs to know where Elasticsearch lives and which Actuator endpoints to expose publicly.

  • Open src/main/resources/application.properties and add these lines:
# Elasticsearch
spring.elasticsearch.uris=http://localhost:9200

# Actuator endpoints
management.endpoints.web.exposure.include=health,info,prometheus,metrics
management.endpoint.health.show-details=always

What do these properties do?

The spring.elasticsearch.uris property tells the Elasticsearch REST client where to connect. The management.endpoints.web.exposure.include line exposes the Prometheus metrics endpoint, which is not exposed by default.

  • Save application.properties.
  • Start your Spring Boot application by running:
mvn spring-boot:run

Wait for the application to start. You should see the Spring Boot banner and log output indicating the app has started on port 8080.

Before you check the metrics endpoint, what do you think you will see at /actuator/prometheus? Plain text? JSON? Something else?

  • Verify the Prometheus metrics endpoint by running this in a new terminal:
curl http://localhost:8080/actuator/prometheus | head -20

You should see plain-text metrics lines in Prometheus exposition format, with entries like jvm_memory_used_bytes and http_server_requests_seconds. This confirms that Prometheus can scrape your application.

  • Open http://localhost:9090/targets in your browser.

You should see the threat-pipeline target with state UP. This confirms Prometheus is successfully scraping metrics from your running application.

Prometheus target still shows DOWN?

Make sure your Spring Boot app is running on port 8080 on the host. The host.docker.internal hostname only works on Docker Desktop for macOS and Windows.

If you are on an older version of Docker Desktop, add this to the prometheus service in your docker-compose.yml:

    extra_hosts:
      - "host.docker.internal:host-gateway"

Then run docker compose up -d prometheus to recreate the container.

Help me fix my Prometheus scraping issue

✔️ Awesome, I've got everything!

Great. Confirm application.properties is saved.

ⓧ I'd like to double check the full code

Your new additions to application.properties should include:

# Elasticsearch
spring.elasticsearch.uris=http://localhost:9200

# Actuator endpoints
management.endpoints.web.exposure.include=health,info,prometheus,metrics
management.endpoint.health.show-details=always

Your observability stack is live. Elasticsearch is ready to index documents, Kibana is waiting for data to explore, and Prometheus is already collecting JVM and HTTP metrics from your application. Next up, you will index your threat events into Elasticsearch so you can search them.

Index Threat Data into Elasticsearch

Your threat events and alerts currently live in PostgreSQL. PostgreSQL is excellent for transactional writes, but it struggles with full-text search across multiple fields simultaneously.

In this step, you will create Elasticsearch document mappings for your existing data, build repository interfaces to interact with those indices, and wire up a service that syncs processed events into Elasticsearch. By the end, you will see your indexed data appear in Kibana.

In this step, get ready to:
  • Create Elasticsearch document classes that map your JPA entity fields to search-optimized indices.
  • Create Spring Data Elasticsearch repository interfaces for indexing operations.
  • Build an indexing service that saves threat data into Elasticsearch and verify documents appear in Kibana.
Create the ThreatEventDocument class

Elasticsearch stores data as JSON documents inside indices. To tell Spring Data how to map your Java objects to those documents, you annotate a class with @Document and mark each field with @Field.

  • Create a new package called com.threatpipeline.search in your src/main/java directory.
  • Inside that package, create a new file called ThreatEventDocument.java and paste the following code:
package com.threatpipeline.search;

import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;
import org.springframework.data.elasticsearch.annotations.DateFormat;

import java.time.Instant;

@Document(indexName = "threat-events")
public class ThreatEventDocument {

    @Id
    private String id;

    @Field(type = FieldType.Text)
    private String eventType;

    @Field(type = FieldType.Keyword)
    private String sourceIp;

    @Field(type = FieldType.Text)
    private String username;

    @Field(type = FieldType.Text)
    private String description;

    @Field(type = FieldType.Keyword)
    private String processingStatus;

What do these annotations mean?

The @Document(indexName = "threat-events") annotation tells Spring Data to map this class to an Elasticsearch index called threat-events.

Each @Field annotation controls how Elasticsearch indexes and searches that field. FieldType.Text enables full-text search (tokenized, analyzed). FieldType.Keyword enables exact-match filtering (no tokenization).

  • Add the remaining fields and the no-arg constructor directly below processingStatus in the same file:
    @Field(type = FieldType.Date, format = DateFormat.date_time)
    private Instant eventTimestamp;

    @Field(type = FieldType.Date, format = DateFormat.date_time)
    private Instant indexedAt;

    public ThreatEventDocument() {}

    // Getters and setters
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }
    public String getEventType() { return eventType; }
    public void setEventType(String eventType) { this.eventType = eventType; }
    public String getSourceIp() { return sourceIp; }
    public void setSourceIp(String sourceIp) { this.sourceIp = sourceIp; }
    public String getUsername() { return username; }
    public void setUsername(String username) { this.username = username; }
    public String getDescription() { return description; }
    public void setDescription(String description) { this.description = description; }
    public String getProcessingStatus() { return processingStatus; }
    public void setProcessingStatus(String processingStatus) { this.processingStatus = processingStatus; }
    public Instant getEventTimestamp() { return eventTimestamp; }
    public void setEventTimestamp(Instant eventTimestamp) { this.eventTimestamp = eventTimestamp; }
    public Instant getIndexedAt() { return indexedAt; }
    public void setIndexedAt(Instant indexedAt) { this.indexedAt = indexedAt; }
}

Why Keyword vs Text for sourceIp?

IP addresses like 192.168.1.100 are exact values you filter on, not words you search through. Using Keyword means Elasticsearch stores the IP as-is for exact matching. If you used Text, Elasticsearch would tokenize it into fragments like 192, 168, 1, 100 and break your filters.

  • Save ThreatEventDocument.java.
Create the ThreatAlertDocument class
  • In the same com.threatpipeline.search package, create a new file called ThreatAlertDocument.java and paste the field declarations:
package com.threatpipeline.search;

import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;
import org.springframework.data.elasticsearch.annotations.DateFormat;

import java.time.Instant;

@Document(indexName = "threat-alerts")
public class ThreatAlertDocument {

    @Id
    private String id;

    @Field(type = FieldType.Keyword)
    private String severity;

    @Field(type = FieldType.Text)
    private String reason;

    @Field(type = FieldType.Keyword)
    private String sourceIp;

    @Field(type = FieldType.Keyword)
    private String relatedEventId;

    @Field(type = FieldType.Date, format = DateFormat.date_time)
    private Instant createdAt;
  • Add the constructor and getters/setters below the createdAt field:
    public ThreatAlertDocument() {}

    // Getters and setters
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }
    public String getSeverity() { return severity; }
    public void setSeverity(String severity) { this.severity = severity; }
    public String getReason() { return reason; }
    public void setReason(String reason) { this.reason = reason; }
    public String getSourceIp() { return sourceIp; }
    public void setSourceIp(String sourceIp) { this.sourceIp = sourceIp; }
    public String getRelatedEventId() { return relatedEventId; }
    public void setRelatedEventId(String relatedEventId) { this.relatedEventId = relatedEventId; }
    public Instant getCreatedAt() { return createdAt; }
    public void setCreatedAt(Instant createdAt) { this.createdAt = createdAt; }
}
  • Save ThreatAlertDocument.java.
Create the repository interfaces and verify index creation

Spring Data Elasticsearch repositories provide built-in save(), findAll(), and delete() operations. When the app starts, Spring Data automatically creates the Elasticsearch indices defined by your @Document classes.

  • In the com.threatpipeline.search package, create ThreatEventSearchRepository.java with the following code:
package com.threatpipeline.search;

import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;

// Spring Data creates CRUD operations automatically for ThreatEventDocument
public interface ThreatEventSearchRepository extends ElasticsearchRepository<ThreatEventDocument, String> {
}
  • In the same package, create ThreatAlertSearchRepository.java:
package com.threatpipeline.search;

import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;

// Spring Data creates CRUD operations automatically for ThreatAlertDocument
public interface ThreatAlertSearchRepository extends ElasticsearchRepository<ThreatAlertDocument, String> {
}

What does ElasticsearchRepository give you?

By extending ElasticsearchRepository, your interface inherits methods like save(), findById(), and deleteAll() without writing any implementation. Spring Data generates the implementation at startup and auto-creates the corresponding Elasticsearch index.

  • Save both repository files.
  • Start your Spring Boot application by running:
mvn spring-boot:run

Before you check the result, what do you think happened in Elasticsearch when the app started?

  • Verify that Spring Data created the indices by running:
curl http://localhost:9200/_cat/indices?v

You should see two indices listed: threat-events and threat-alerts. Both will show a document count of 0, since you have not indexed any data yet.

Not seeing the indices?

  • Check the Spring Boot console output for errors. If you see a connection refused message, confirm Elasticsearch is running with docker ps and that the elasticsearch container shows as healthy.
  • If you see a mapping error mentioning your @Field annotations, double check that sourceIp and severity use FieldType.Keyword and that description and username use FieldType.Text.
  • If the index exists with an old mapping from a previous attempt, delete it and restart: run curl -X DELETE http://localhost:9200/threat-events then restart the app.

Help me troubleshoot my Elasticsearch index creation with Spring Data

  • Stop the application with Ctrl+C in the terminal.
Create the ElasticsearchIndexService

The document classes and repositories define the shape and storage of your search data. Now you need a service that takes threat event data and saves it into Elasticsearch using those repositories.

  • In the com.threatpipeline.search package, create ElasticsearchIndexService.java and paste the class declaration with its constructor:
package com.threatpipeline.search;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.time.Instant;

@Service
public class ElasticsearchIndexService {

    private static final Logger log = LoggerFactory.getLogger(ElasticsearchIndexService.class);

    private final ThreatEventSearchRepository eventSearchRepository;
    private final ThreatAlertSearchRepository alertSearchRepository;

    // Spring injects the repositories automatically
    public ElasticsearchIndexService(ThreatEventSearchRepository eventSearchRepository,
                                     ThreatAlertSearchRepository alertSearchRepository) {
        this.eventSearchRepository = eventSearchRepository;
        this.alertSearchRepository = alertSearchRepository;
    }
}

What does this class set up?

The @Service annotation registers this class as a Spring bean. The constructor injection pattern lets Spring automatically provide both repository instances. The Logger gives you visibility into which documents get indexed.

  • Add the indexThreatEvent method inside the class, below the constructor and before the closing }:
    public void indexThreatEvent(Long eventId, String eventType, String sourceIp,
                                  String username, String description,
                                  String processingStatus, Instant eventTimestamp) {
        // Build a document from the raw event data
        ThreatEventDocument doc = new ThreatEventDocument();
        doc.setId(eventId.toString());
        doc.setEventType(eventType);
        doc.setSourceIp(sourceIp);
        doc.setUsername(username);
        doc.setDescription(description);
        doc.setProcessingStatus(processingStatus);
        doc.setEventTimestamp(eventTimestamp);
        doc.setIndexedAt(Instant.now());

        eventSearchRepository.save(doc);
        log.info("Indexed threat event id={} into Elasticsearch", eventId);
    }

What does indexThreatEvent do?

This method takes the raw fields from a processed threat event, builds a ThreatEventDocument, and saves it to Elasticsearch via the repository. The indexedAt timestamp records when the document entered the search index. The log line confirms each successful index operation.

  • Add the indexThreatAlert method directly below indexThreatEvent:
    public void indexThreatAlert(Long alertId, String severity, String reason,
                                  String sourceIp, Long relatedEventId, Instant createdAt) {
        // Build a document from the alert data
        ThreatAlertDocument doc = new ThreatAlertDocument();
        doc.setId(alertId.toString());
        doc.setSeverity(severity);
        doc.setReason(reason);
        doc.setSourceIp(sourceIp);
        doc.setRelatedEventId(relatedEventId.toString());
        doc.setCreatedAt(createdAt);

        alertSearchRepository.save(doc);
        log.info("Indexed threat alert id={} severity={} into Elasticsearch", alertId, severity);
    }

What does indexThreatAlert do?

This follows the same pattern as indexThreatEvent but for alert documents. The relatedEventId field links each alert back to the threat event that triggered it, making it possible to search for all alerts related to a specific event.

  • Save ElasticsearchIndexService.java.
  • Start your application again to confirm everything compiles:
mvn spring-boot:run

The app should start without errors. You will see your existing pipeline start up alongside the new Elasticsearch components.

Seeing a bean creation error?

  • If the error mentions ThreatEventSearchRepository or ThreatAlertSearchRepository, confirm both interface files are in the com.threatpipeline.search package and extend ElasticsearchRepository.
  • If the error mentions a missing ElasticsearchOperations bean, confirm spring.elasticsearch.uris=http://localhost:9200 is in your application.properties and Elasticsearch is running.

Help me fix my Spring Boot startup error related to Elasticsearch beans

✔️ Awesome, I've got everything!

Great. Double check you have saved all five files in the com.threatpipeline.search package.

ⓧ I'd like to double check the full code

Confirm each file is present and correct by scrolling up to the code chunks you pasted. You should have these five files in src/main/java/com/threatpipeline/search/:

  • ThreatEventDocument.java with 8 annotated fields, a no-arg constructor, and getters/setters.
  • ThreatAlertDocument.java with 5 annotated fields, a no-arg constructor, and getters/setters.
  • ThreatEventSearchRepository.java extending ElasticsearchRepository<ThreatEventDocument, String>.
  • ThreatAlertSearchRepository.java extending ElasticsearchRepository<ThreatAlertDocument, String>.
  • ElasticsearchIndexService.java with indexThreatEvent() and indexThreatAlert() methods.
Verify a document appears in Kibana Dev Tools

Your indexing service is ready, but the Elasticsearch indices are still empty. To prove the full pipeline works, you will POST a threat event through your existing REST API, call the indexing service to sync it, and then query Elasticsearch from Kibana.

  • With the application still running, open a second terminal and submit a test threat event using your existing POST endpoint:
curl -X POST http://localhost:8080/api/v1/events \
  -H "Content-Type: application/json" \
  -d '{
    "eventType": "LOGIN_FAILURE",
    "sourceIp": "192.168.1.100",
    "username": "admin",
    "description": "Brute force login attempt detected"
  }'

What happens when you POST this event?

Your existing pipeline from Project 1 receives the event, stores it in PostgreSQL, publishes it to Kafka, and processes it through your alert rules. The ElasticsearchIndexService hooks into this flow to also save the event into Elasticsearch. Check the application logs for the Indexed threat event id= message confirming the sync.

What do you think Elasticsearch contains now? Let's check.

  • Open Kibana Dev Tools in your browser at http://localhost:5601/app/dev_tools.
  • In the Dev Tools console, run this query:
GET /threat-events/_search

You should see a JSON response with "hits" containing your indexed document. The _source field shows the stored event data including eventType, sourceIp, username, and description.

Seeing zero hits in the search result?

  • Check your application logs for the Indexed threat event log line. If it is missing, your event processing flow is not calling ElasticsearchIndexService.indexThreatEvent(). Confirm you wired the service into your existing Kafka consumer or event processor.
  • If you see a log line but no hits, Elasticsearch may not have refreshed yet. Run POST /threat-events/_refresh in Kibana Dev Tools, then retry the search.
  • If the POST to your events endpoint returned an error, confirm your existing Project 1 services (PostgreSQL, Kafka) are running with docker ps.

Help me debug why my threat event is not appearing in Elasticsearch

Your threat data is flowing into Elasticsearch and you can see it in Kibana. Next up, you will build a search API that lets you query this data with full-text search and filtering.

Build the Full-Text Search API

Your threat events are now indexed in Elasticsearch, but they are sitting there with no way to query them programmatically. A security analyst investigating a brute force attack needs to search across descriptions, usernames, and source IPs simultaneously, then narrow results by severity or time window.

In this step, you will build the search service and REST controller that make this possible. The service constructs dynamic Elasticsearch queries combining full-text search with structured filters, and a controller exposes it all through a clean HTTP endpoint.

In this step, get ready to:
  • Create a search service that builds bool queries with full-text and filter clauses.
  • Create a REST controller exposing a paginated search endpoint.
  • Test the API with keyword, filter, and combined queries and save a search in Kibana.
Create the search service

The search service is where the query logic lives. It uses a BoolQuery to combine a full-text multi_match clause with optional term, match, and range filters depending on which parameters the caller provides.

  • Create a new file called ThreatSearchService.java in src/main/java/com/threatpipeline/search/ and add the class skeleton with its constructor:
package com.threatpipeline.search;

import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.client.elc.NativeQuery;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.util.List;

@Service
public class ThreatSearchService {

    private final ElasticsearchOperations elasticsearchOperations;
    private final Timer searchTimer;

    // Registers a custom Micrometer timer to track ES query duration
    public ThreatSearchService(ElasticsearchOperations elasticsearchOperations,
                                MeterRegistry meterRegistry) {
        this.elasticsearchOperations = elasticsearchOperations;
        this.searchTimer = Timer.builder("threat.search.query.duration")
                .description("Time taken to execute Elasticsearch search queries")
                .register(meterRegistry);
    }
}

What does this code do?

The constructor injects two Spring-managed beans. ElasticsearchOperations is the Spring Data Elasticsearch client that executes queries against your cluster.

MeterRegistry comes from Micrometer and lets you register custom metrics. The Timer named threat.search.query.duration tracks how long each Elasticsearch search takes, separately from the HTTP request overhead.

Now add the searchEvents method inside the class. This first block handles the method signature, the full-text clause, and the severity and event type filters.

  • Inside ThreatSearchService, add the following method above the closing } brace:
    // Builds a dynamic bool query combining full-text search with filters
    public SearchHits<ThreatEventDocument> searchEvents(String query, String severity,
                                                         String eventType, String sourceIp,
                                                         Instant from, Instant to,
                                                         int page, int size) {
        return searchTimer.record(() -> {
            BoolQuery.Builder boolBuilder = new BoolQuery.Builder();

            // Full-text search across multiple fields
            if (query != null && !query.isBlank()) {
                boolBuilder.must(Query.of(q -> q.multiMatch(m -> m
                        .query(query)
                        .fields("description", "username", "eventType", "sourceIp")
                )));
            }

            // Exact-match filter on processingStatus for severity
            if (severity != null && !severity.isBlank()) {
                boolBuilder.filter(Query.of(q -> q.term(t -> t
                        .field("processingStatus")
                        .value(severity)
                )));
            }

            // Text match filter for event type
            if (eventType != null && !eventType.isBlank()) {
                boolBuilder.filter(Query.of(q -> q.match(m -> m
                        .field("eventType")
                        .query(eventType)
                )));
            }

What is a bool query?

A bool query in Elasticsearch combines multiple clauses. The must clause contributes to relevance scoring (full-text results ranked by how well they match). The filter clauses are yes/no checks that do not affect scoring, which makes them faster.

The multi_match searches across description, username, eventType, and sourceIp simultaneously. A single keyword like "brute force" matches any document containing it in any of those fields.

Continue the method with the source IP filter, date range filter, and query execution.

  • Directly below the event type filter block you just added, paste the remaining method body:
            // Exact-match filter for source IP address
            if (sourceIp != null && !sourceIp.isBlank()) {
                boolBuilder.filter(Query.of(q -> q.term(t -> t
                        .field("sourceIp")
                        .value(sourceIp)
                )));
            }

            // Date range filter on eventTimestamp
            if (from != null || to != null) {
                boolBuilder.filter(Query.of(q -> q.range(r -> r
                        .date(d -> {
                            d.field("eventTimestamp");
                            if (from != null) d.gte(from.toString());
                            if (to != null) d.lte(to.toString());
                            return d;
                        })
                )));
            }

            // Build and execute the paginated query
            NativeQuery searchQuery = NativeQuery.builder()
                    .withQuery(Query.of(q -> q.bool(boolBuilder.build())))
                    .withPageable(PageRequest.of(page, size))
                    .build();

            return elasticsearchOperations.search(searchQuery, ThreatEventDocument.class);
        });
    }

What does the range filter do?

The range filter uses gte (greater than or equal) and lte (less than or equal) to bracket results within a time window. This lets analysts ask "show me all events in the last 24 hours" by passing from and to timestamps.

NativeQuery wraps the final bool query with pagination via PageRequest, so consumers can page through large result sets. The entire execution is wrapped in searchTimer.record() so Prometheus captures query duration automatically.

  • Save ThreatSearchService.java.
  • Restart your Spring Boot application to confirm it compiles without errors.

Seeing a compilation error?

  • Ensure the closing braces match: the method ends with }); (closing the lambda), then } (closing the method), then } (closing the class).
  • Verify the import for co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery resolves. If your IDE shows a red underline, ensure spring-boot-starter-data-elasticsearch is in your pom.xml and run mvn dependency:resolve.

Help me debug my ThreatSearchService compilation error.

✔️ Awesome, I've got everything!

Great. Double check you have saved ThreatSearchService.java and your application restarts cleanly.

ⓧ I'd like to double check the full code

Your ThreatSearchService.java should contain these sections in order:

  • Package declaration and all imports (BoolQuery, Query, MeterRegistry, Timer, PageRequest, NativeQuery, ElasticsearchOperations, SearchHits, Service, Instant, List).
  • The @Service class with two fields: elasticsearchOperations and searchTimer.
  • The constructor registering threat.search.query.duration timer.
  • The searchEvents method with: multi_match clause, severity term filter on processingStatus, eventType match filter, sourceIp term filter, date range filter, and NativeQuery execution.
Create the search controller

The controller exposes your search service as an HTTP endpoint. It accepts optional query parameters and returns a JSON response with the total hit count and paginated results.

  • Create a new file called ThreatSearchController.java in src/main/java/com/threatpipeline/search/ and add the class with its constructor:
package com.threatpipeline.search;

import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@RestController
@RequestMapping("/api/v1/threats")
public class ThreatSearchController {

    private final ThreatSearchService searchService;

    // Spring injects the ThreatSearchService bean
    public ThreatSearchController(ThreatSearchService searchService) {
        this.searchService = searchService;
    }
}

Why /api/v1/threats?

The @RequestMapping("/api/v1/threats") sets the base path for this controller. The search endpoint will live at /api/v1/threats/search. Versioning the API path (v1) lets you evolve the contract later without breaking existing clients.

  • Now add the search endpoint method inside the class, above the closing } brace:
    // GET endpoint with optional query, filter, and pagination parameters
    @GetMapping("/search")
    public ResponseEntity<Map<String, Object>> search(
            @RequestParam(required = false) String query,
            @RequestParam(required = false) String severity,
            @RequestParam(required = false) String eventType,
            @RequestParam(required = false) String sourceIp,
            @RequestParam(required = false) Instant from,
            @RequestParam(required = false) Instant to,
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "20") int size) {

        SearchHits<ThreatEventDocument> hits = searchService.searchEvents(
                query, severity, eventType, sourceIp, from, to, page, size);

        // Extract document content from search hit wrappers
        List<ThreatEventDocument> results = hits.getSearchHits().stream()
                .map(SearchHit::getContent)
                .toList();

        // Build a response map with metadata and results
        Map<String, Object> response = new HashMap<>();
        response.put("totalHits", hits.getTotalHits());
        response.put("results", results);
        response.put("page", page);
        response.put("size", size);

        return ResponseEntity.ok(response);
    }

What does this endpoint return?

Every parameter is optional (required = false). Callers can pass any combination of full-text query and filters. If no parameters are provided, the endpoint returns all indexed events.

The response JSON includes totalHits (how many documents matched), results (the current page of documents), page, and size for pagination metadata.

  • Save ThreatSearchController.java.
  • Restart your Spring Boot application.

Before you test, do you think a request to the search endpoint with no parameters will return results or an error?

  • Test the empty-query case by running:
curl http://localhost:8080/api/v1/threats/search

You should see a JSON response with totalHits showing the count of all indexed threat events, and the results array containing the test event you indexed in the previous step.

Getting a 404 or 500 error?

  • A 404 means the controller is not being scanned. Confirm ThreatSearchController.java is in the com.threatpipeline.search package and the class has @RestController and @RequestMapping("/api/v1/threats").
  • A 500 with "Elasticsearch connection refused" means the ES container stopped. Check with docker ps and restart with docker compose up -d elasticsearch if needed.

Help me debug my search endpoint error.

✔️ Awesome, I've got everything!

Great. Confirm you have saved ThreatSearchController.java and received a valid JSON response from the endpoint.

ⓧ I'd like to double check the full code

Your ThreatSearchController.java should contain these sections in order:

  • Package declaration and imports (SearchHit, SearchHits, ResponseEntity, GetMapping, RequestMapping, RequestParam, RestController, Instant, HashMap, List, Map).
  • The @RestController class with @RequestMapping("/api/v1/threats") and a constructor injecting ThreatSearchService.
  • The @GetMapping("/search") method accepting query, severity, eventType, sourceIp, from, to, page, size parameters and returning a map with totalHits, results, page, size.
Test the search API and create a Kibana saved search

Your endpoint is live. Now exercise it with several query combinations to prove the bool query logic works for keyword-only, filter-only, and combined searches.

  • Test a keyword-only search by running:
curl "http://localhost:8080/api/v1/threats/search?query=brute+force"

You should see results where the description, username, or source IP contains "brute force".

  • Test a filter-only search (no full-text keyword, just severity) by running:
curl "http://localhost:8080/api/v1/threats/search?severity=PROCESSED"

This returns only events whose processingStatus field exactly matches PROCESSED.

Now for the final verification. What do you think happens when you combine a keyword with a severity filter?

  • Test the combined keyword + filter search by running:
curl "http://localhost:8080/api/v1/threats/search?query=brute+force&severity=HIGH"

You receive a JSON response containing only threat events that match "brute force" in a text field AND have processingStatus set to HIGH. If you only indexed one test event with a different status, you may see zero results here.

Why zero results is still a valid response

If your test event has processingStatus set to PROCESSED rather than HIGH, the filter correctly excludes it. The response shape (totalHits, results, page, size) proves the endpoint is working. The filters are doing their job.

Now create a saved search in Kibana that demonstrates the same filtering capability through the UI.

  • Open Kibana at http://localhost:5601.
  • Click **Discover** in the navigation menu.
  • Create a data view pointing to the threat-events index if prompted. Set the timestamp field to eventTimestamp.
  • In the search bar, type a KQL query such as description: "brute force" and press **Enter** to filter results.
  • In the application menu bar, click **Save**.
  • Enter High Severity Threat Search as the title.
  • Click **Save** to store it.

Your saved Discover session now preserves the query, filters, and column configuration so you can reload it later during threat investigations.

Your full-text search API is live and filtering correctly. Next up, you will wire Prometheus and Grafana to monitor how fast these searches run and whether errors are creeping in.

Add Metrics and Grafana Dashboards

Your search API returns results, but you have no idea how fast it responds, how often it errors, or whether the JVM is under memory pressure. Without metrics, you are flying blind in production.

When a security analyst reports "search is slow," you need data to diagnose the problem. In this step, you will verify that your application exposes metrics, confirm Prometheus is collecting them, and build a Grafana dashboard that visualizes your search API's health in real time.

In this step, get ready to:
  • Verify the Prometheus metrics endpoint exposes search-related metrics.
  • Confirm Prometheus is scraping your Spring Boot application successfully.
  • Build a Grafana dashboard with panels for request rate, latency, errors, and JVM health.
Verify the Prometheus metrics endpoint

The micrometer-registry-prometheus dependency you added in Step 1 auto-registers a Prometheus MeterRegistry with Spring Boot. This means every HTTP request your application handles is already being measured and exposed at the /actuator/prometheus endpoint.

  • With your Spring Boot application running, open a new terminal and check the metrics endpoint by running this command:
curl http://localhost:8080/actuator/prometheus | grep http_server_requests_seconds

What does this command do?

This fetches all Prometheus-format metrics from your app and filters for http_server_requests_seconds lines. These are the default HTTP server metrics that Micrometer records for every request to your application.

You should see multiple lines with metric names like http_server_requests_seconds_count and http_server_requests_seconds_bucket in the output.

  • Now confirm your custom timer is also exposed by running:
curl http://localhost:8080/actuator/prometheus | grep threat.search.query.duration

You should see threat_search_query_duration_seconds metrics in the output. Prometheus converts dots to underscores in metric names.

Why two sets of metrics?

The http_server_requests_seconds metric measures the full HTTP request lifecycle, including serialization and network overhead.

The custom threat.search.query.duration timer in ThreatSearchService measures only the Elasticsearch query execution time. This separation lets you distinguish between slow queries and slow network responses.

Not seeing any metrics?

Make sure your Spring Boot application is running on port 8080. If curl returns connection refused, start the application first.

If you see metrics but not the http_server_requests_seconds lines, confirm that management.endpoints.web.exposure.include in application.properties includes prometheus.

help me debug why the /actuator/prometheus endpoint is not returning metrics

Confirm Prometheus scraping and generate metric data

Your prometheus.yml configuration file tells Prometheus to scrape your Spring Boot app every 15 seconds. Now you need to verify that Prometheus has connected successfully.

  • Open the Prometheus targets page in your browser at http://localhost:9090/targets.

You should see a target named threat-pipeline with the state showing UP in green. This confirms Prometheus is successfully pulling metrics from your application.

Target showing DOWN?

Prometheus runs inside Docker and uses host.docker.internal to reach your Spring Boot app on the host machine. If the target shows DOWN, confirm your Spring Boot app is actually running on port 8080 on the host.

On older Docker Desktop versions, you may need to add extra_hosts: ["host.docker.internal:host-gateway"] to the prometheus service in docker-compose.yml.

help me fix Prometheus target showing DOWN for my Spring Boot application

Now generate some metric data by running several search queries against your API.

  • Execute a series of search queries by running these commands:
# Full-text search
curl "http://localhost:8080/api/v1/threats/search?query=brute+force"

# Filtered search by severity
curl "http://localhost:8080/api/v1/threats/search?severity=HIGH"

# Combined search with multiple filters
curl "http://localhost:8080/api/v1/threats/search?query=login&severity=HIGH&eventType=LOGIN_FAILURE"

Why run multiple queries?

Prometheus scrapes your app every 15 seconds, but rate and histogram calculations need multiple data points to produce meaningful results. Running several queries gives Prometheus enough samples to display non-zero values on your Grafana panels.

  • Verify Prometheus collected the data by opening http://localhost:9090 in your browser.
  • Type http_server_requests_seconds_bucket into the query field and click **Execute**.

You should see time-series results with labels including uri="/api/v1/threats/search" confirming Prometheus has your search endpoint data.

Build a Grafana monitoring dashboard

With metric data flowing into Prometheus, you can now visualize it in Grafana. You will create a dashboard with four panels that cover the key observability signals for your search API.

  • Open Grafana in your browser at http://localhost:3000.
  • Log in with username admin and password admin.
  • Grafana will prompt you to change the password. Click **Skip** for now.

First, connect Grafana to Prometheus as a data source.

  • Click **Connections** in the left-side menu.
  • Click **Add new connection**.
  • Type Prometheus in the search bar and select the **Prometheus** data source.
  • Click **Add new data source** in the upper right.
  • In the **Connection** section, enter http://prometheus:9090 in the **Prometheus server URL** field.
  • Scroll to the bottom and click **Save & test**.

You should see a green success message confirming the data source is working.

Why prometheus:9090 instead of localhost?

Grafana runs inside a Docker container on the same Docker network as Prometheus. The hostname prometheus resolves to the Prometheus container's IP address within the threat-network Docker network. Using localhost would point Grafana at its own container, not Prometheus.

Save & test failing?

Make sure the URL is exactly http://prometheus:9090 with no trailing slash. Confirm the Prometheus container is running by checking docker ps lists a container named prometheus.

help me connect Grafana to Prometheus as a data source

Now create your monitoring dashboard.

  • Click **Dashboards** in the left-side menu.
  • Click **New** and select **New Dashboard**.
  • Click the **Add new element** icon (blue plus sign) and drag a panel onto the dashboard.
  • On the new panel, click **Configure visualization**.

The panel editor opens with Prometheus pre-selected as the data source. You will create the first panel: search request rate.

  • In the query editor at the bottom, switch to **Code** mode if the builder view is showing.
  • Enter the following PromQL query:
rate(http_server_requests_seconds_count{uri="/api/v1/threats/search"}[5m])
  • Click **Refresh** to execute the query.
  • Select **Time series** as the visualization type.
  • In the panel title field, enter Search Request Rate.
  • Click **Save** to save the panel.

You should see a time-series graph showing the rate of requests to your search endpoint over the last 5 minutes.

  • Click **Back to dashboard** to return to the dashboard view.
  • Click the **Add new element** icon and drag another panel onto the dashboard.
  • Click **Configure visualization** on the new panel.
  • Enter this PromQL query for P95 latency:
histogram_quantile(0.95, rate(http_server_requests_seconds_bucket{uri="/api/v1/threats/search"}[5m]))
  • Click **Refresh**, select **Time series**, and set the title to P95 Search Latency.
  • Click **Save** and then **Back to dashboard**.
  • Add a third panel with this query for error count by status code:
rate(http_server_requests_seconds_count{uri="/api/v1/threats/search", status!="200"}[5m])
  • Click **Refresh**, select **Time series**, and set the title to Error Count by Status.
  • Click **Save** and then **Back to dashboard**.
  • Add a fourth panel with this query for JVM heap usage:
jvm_memory_used_bytes{area="heap"}
  • Click **Refresh**, select **Time series**, and set the title to JVM Heap Usage.
  • Click **Save** to save the panel.
  • Click **Back to dashboard**.

You now have a dashboard with four panels showing search request rate, P95 latency, error count, and JVM heap usage.

  • Click **Save** in the top toolbar.
  • Enter Threat Search Monitoring as the dashboard title.
  • Click **Save**.

What do these panels tell you?

The Search Request Rate panel shows how many requests per second your search API handles. Spikes here could indicate automated scanning or a surge in analyst activity.

The P95 Search Latency panel shows the response time for the slowest 5% of requests. If this climbs above a few hundred milliseconds, your Elasticsearch queries may need optimization.

The Error Count panel surfaces any non-200 responses. The JVM Heap Usage panel warns you before the app runs out of memory.

Before you move on, what do you expect to see if you run a few more search queries right now and then refresh the Grafana dashboard?

  • Run a few more search queries to generate fresh data:
curl "http://localhost:8080/api/v1/threats/search?query=port+scan"
curl "http://localhost:8080/api/v1/threats/search?query=malware&severity=CRITICAL"
  • Return to your Grafana dashboard and click the refresh icon in the top-right corner.

You should see the Search Request Rate panel update with new data points reflecting your recent queries. The P95 latency panel shows how quickly Elasticsearch responded to those searches.

Dashboard panels showing No data?

Check that the Prometheus data source is working by clicking the **Explore** link in the left menu, selecting Prometheus, and running a simple query like up. If that returns results, your data source connection is fine.

If individual panels show no data, confirm the URI label matches exactly. Run curl http://localhost:8080/actuator/prometheus | grep uri and check whether the label value is /api/v1/threats/search or a different path.

help me fix Grafana panels showing No data for my Prometheus queries

You now have full observability over your search API. Next up, you will add structured JSON logging and integration tests to make your pipeline production-ready and provably correct.

Add Structured Logging and Integration Tests

Your search API and monitoring dashboards are working, but two gaps remain. Your application logs are plain text, which means no log aggregator can parse them automatically.

In this step, you will configure structured JSON logging with logstash-logback-encoder and write integration tests using Testcontainers. The tests spin up a real Elasticsearch container and verify your search pipeline end-to-end without mocking.

In this step, get ready to:
  • Configure structured JSON logging with a Logback configuration file.
  • Create integration tests using Testcontainers and a real Elasticsearch container.
  • Run the test suite and confirm all tests pass with structured log output.
Configure structured JSON logging

Plain-text logs look fine in a terminal, but they are impossible for log aggregators like the ELK stack, Datadog, or CloudWatch to parse without custom regex rules. LogstashEncoder from logstash-logback-encoder produces JSON logs with standard fields that any aggregator can ingest immediately.

  • Create a new file called logback-spring.xml inside src/main/resources/ (the same folder as your application.properties).
  • Paste the following configuration into the file:
<?xml version="1.0" encoding="UTF-8"?>
<configuration>

    <springProfile name="json">
        <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
            <encoder class="net.logstash.logback.encoder.LogstashEncoder">
                <customFields>{"application":"threat-pipeline","environment":"local"}</customFields>
            </encoder>
        </appender>
    </springProfile>

    <springProfile name="!json">
        <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
            <encoder>
                <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
            </encoder>
        </appender>
    </springProfile>

    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
    </root>

    <logger name="com.threatpipeline" level="DEBUG"/>
    <logger name="org.springframework.data.elasticsearch" level="WARN"/>
</configuration>

What does this configuration do?

The <springProfile name="json"> block activates only when you run with the json profile. It uses LogstashEncoder to emit each log line as a JSON object with fields like @timestamp, level, logger_name, thread_name, and message.

The <springProfile name="!json"> block is the fallback for local development. It keeps the familiar human-readable format so your console stays clean during coding.

The custom loggers suppress noisy Elasticsearch client output while giving your own com.threatpipeline package DEBUG visibility.

  • Save logback-spring.xml.
  • Stop your running application if it is still active, then restart it with the json profile by running:
mvn spring-boot:run -Dspring-boot.run.profiles=json

You should see every log line printed as a JSON object in your terminal. Each line will include fields like "level":"INFO" and "logger_name":"..." instead of the plain-text format you saw before.

Still seeing plain-text logs?

Make sure the -Dspring-boot.run.profiles=json flag is present in your command. Without it, the !json profile activates and uses the plain-text encoder.

Also confirm your logback-spring.xml file is saved inside src/main/resources/ (not a subfolder). Spring Boot auto-detects this file by name.

help me debug my structured logging setup

  • Stop the application with Ctrl+C once you have confirmed JSON output.
Create the integration test class

Testcontainers lets your test suite spin up a real Elasticsearch Docker container, run queries against it, and tear it down automatically. No mocking, no relying on your Docker Compose stack being up.

  • Create a new file at src/test/java/com/threatpipeline/search/ThreatSearchIntegrationTest.java.
  • Start with the class structure, imports, and the Testcontainers container definition by pasting the following:
package com.threatpipeline.search;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.elasticsearch.ElasticsearchContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

import java.time.Instant;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Testcontainers
class ThreatSearchIntegrationTest {

    @Container
    static ElasticsearchContainer elasticsearch = new ElasticsearchContainer(
            "docker.elastic.co/elasticsearch/elasticsearch:8.18.0")
            .withEnv("xpack.security.enabled", "false")
            .withEnv("discovery.type", "single-node");

What does this setup do?

The @Testcontainers annotation tells JUnit to manage container lifecycle. The @Container field creates a real Elasticsearch 8.18.0 container with security disabled, matching your Docker Compose configuration.

Because the container maps to a random port, you will use @DynamicPropertySource next to inject the container's actual address into Spring's configuration at runtime.

  • Below the container field, add the dynamic property source, autowired fields, and setup method:
    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.elasticsearch.uris", elasticsearch::getHttpHostAddress);
    }

    @Autowired
    private ThreatEventSearchRepository eventSearchRepository;

    @Autowired
    private ThreatSearchService searchService;

    @BeforeEach
    void setUp() {
        eventSearchRepository.deleteAll();
    }

What does this wiring accomplish?

The @DynamicPropertySource method overrides spring.elasticsearch.uris with the container's actual host and port. This means your tests connect to the Testcontainers instance instead of your Docker Compose Elasticsearch.

The @BeforeEach method calls deleteAll() so each test starts with a clean index. This prevents test pollution where one test's data affects another.

  • Below the setUp() method, add the first test that verifies full-text keyword search:
    @Test
    void shouldIndexAndSearchByKeyword() {
        ThreatEventDocument doc = new ThreatEventDocument();
        doc.setId("1");
        doc.setEventType("LOGIN_FAILURE");
        doc.setSourceIp("192.168.1.100");
        doc.setUsername("admin");
        doc.setDescription("Brute force login attempt detected");
        doc.setProcessingStatus("PROCESSED");
        doc.setEventTimestamp(Instant.now());
        doc.setIndexedAt(Instant.now());
        eventSearchRepository.save(doc);

        // Elasticsearch needs a brief refresh
        eventSearchRepository.findAll();

        SearchHits<ThreatEventDocument> results = searchService.searchEvents(
                "brute force", null, null, null, null, null, 0, 10);

        assertThat(results.getTotalHits()).isGreaterThanOrEqualTo(1);
        assertThat(results.getSearchHits().get(0).getContent().getDescription())
                .contains("Brute force");
    }

What does this test prove?

This test indexes a single document describing a brute force attack, then searches for "brute force" using your ThreatSearchService. If the full-text multi-match query works correctly, the result contains the indexed document.

The eventSearchRepository.findAll() call forces Elasticsearch to refresh its index so the document becomes searchable immediately.

  • Below the first test method, add the second test that sets up the test data for severity filtering:
    @Test
    void shouldFilterBySeverity() {
        ThreatEventDocument doc1 = new ThreatEventDocument();
        doc1.setId("2");
        doc1.setEventType("PORT_SCAN");
        doc1.setSourceIp("10.0.0.50");
        doc1.setUsername("scanner");
        doc1.setDescription("Port scanning activity");
        doc1.setProcessingStatus("HIGH");
        doc1.setEventTimestamp(Instant.now());
        doc1.setIndexedAt(Instant.now());

        ThreatEventDocument doc2 = new ThreatEventDocument();
        doc2.setId("3");
        doc2.setEventType("INFO_DISCLOSURE");
        doc2.setSourceIp("10.0.0.51");
        doc2.setUsername("user1");
        doc2.setDescription("Information disclosure attempt");
        doc2.setProcessingStatus("LOW");
        doc2.setEventTimestamp(Instant.now());
        doc2.setIndexedAt(Instant.now());
  • Directly below the doc2 setup lines, add the save calls, the search execution, assertions, and close both the method and class:
        eventSearchRepository.save(doc1);
        eventSearchRepository.save(doc2);
        eventSearchRepository.findAll();

        SearchHits<ThreatEventDocument> results = searchService.searchEvents(
                null, "HIGH", null, null, null, null, 0, 10);

        assertThat(results.getTotalHits()).isEqualTo(1);
        assertThat(results.getSearchHits().get(0).getContent().getSourceIp())
                .isEqualTo("10.0.0.50");
    }
}

What does the severity filter test prove?

This test indexes two documents with different processingStatus values (HIGH and LOW). It then searches with a severity filter of HIGH and no keyword.

Only the HIGH document should come back. This proves the term filter in your BoolQuery correctly narrows results by structured fields.

  • Save ThreatSearchIntegrationTest.java.

Seeing a compilation error?

If your IDE reports an unresolved import for ElasticsearchContainer, confirm that testcontainers-elasticsearch is in your pom.xml with <scope>test</scope>. Reload your Maven project after checking.

help me fix Testcontainers import errors in my integration test

Run the test suite and verify

Your integration test is ready. Before you run it, think about what you expect to happen.

Do you think both tests will pass on the first run? What will Testcontainers do behind the scenes before your assertions execute?

  • Run the full test suite:
mvn test

Testcontainers pulls the Elasticsearch 8.18.0 image (if not cached), starts a container, waits for the health check, injects the address into your Spring context, and then runs your two tests. Both tests should pass with green output.

You should also see structured JSON log lines in the console output during the test run, proving the logback-spring.xml encoder is active.

Tests failing with connection refused?

Ensure Docker Desktop is running. Testcontainers needs Docker to start the Elasticsearch container.

If you see a mapper_parsing_exception, it usually means an existing index has a conflicting mapping. Since Testcontainers creates a fresh container each run, this should not happen unless you are reusing containers.

help me debug my Testcontainers Elasticsearch integration test failures

✔️ Awesome, I've got everything!

Great. Double check you have saved both logback-spring.xml and ThreatSearchIntegrationTest.java.

ⓧ I'd like to double check the full code

Scroll up and confirm each chunk is present in order. Your logback-spring.xml should be the complete file from the first chunk.

Your ThreatSearchIntegrationTest.java should contain all five chunks combined: the class structure (imports through container definition), the dynamic property source and fields, the keyword search test, the severity filter test data setup, and the save/assert/close block. Make sure the class opens with class ThreatSearchIntegrationTest { and closes with a final } after the second test method's closing brace.

Your threat search pipeline now has machine-parseable JSON logs and automated tests that prove indexing and search work end-to-end. The next challenge is to add distributed tracing so you can follow a single request across your entire stack.

Secret mission

Distributed Tracing with Zipkin

Your search API works and your metrics dashboard is live. But when a security analyst reports a slow search, can you pinpoint exactly which part of the request took the longest? Add distributed tracing with Zipkin to visualize the full lifecycle of every request, from HTTP handling through Elasticsearch query execution.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely on local Docker containers, so there are no ongoing financial costs. The containers do consume approximately 3-4 GB of RAM while running.

Resources you used:

  • Docker containers: Elasticsearch, Kibana, Prometheus, Grafana, and Zipkin.
  • Docker volumes: esdata, prometheus_data, and grafana_storage.
  • Grafana dashboard and Prometheus data source configuration (stored inside the grafana_storage volume).
  • Elasticsearch indices: threat-events and threat-alerts (stored inside the esdata volume).
  • Local project files: new Java source files in src/main/java/com/threatpipeline/search/, test files, and configuration files (prometheus.yml, logback-spring.xml).

Keep everything running

No action needed. Choose this if you are still actively building on the threat pipeline or plan to continue with future projects in this series.

  • Your Docker containers will continue running and consuming RAM (approximately 3-4 GB total).
  • Your Elasticsearch indices, Grafana dashboards, and Prometheus metrics history will persist across restarts.
  • Your Spring Boot application can be stopped and restarted at any time without affecting the containers.

Pause - I'll come back to this later

Shut down running containers to free up memory while keeping all data intact. Resume later without losing anything.

  • Stop your Spring Boot application if it is still running (press Ctrl+C in the terminal where it runs).
  • Stop all Docker containers by running:
docker compose stop
  • When you are ready to resume, start everything back up by running:
docker compose start

All your volumes, indices, dashboards, and metrics history will be exactly where you left them.

Delete - I don't want to use this again

Remove all containers and volumes created by this project. This frees all disk space and memory but permanently deletes your Elasticsearch indices, Grafana dashboards, and Prometheus metrics history.

  • Stop your Spring Boot application if it is still running (press Ctrl+C in the terminal).
  • Remove all containers and volumes by running:
docker compose down -v
  • Verify the containers are gone by running:
docker ps

You should see no containers related to this project (no elasticsearch, kibana, prometheus, grafana, or zipkin).

  • Verify the volumes are removed by running:
docker volume ls

You should no longer see esdata, prometheus_data, or grafana_storage in the list.

What about the project source files?

The Java source files, test files, and configuration files you created remain in your project directory. These are part of your codebase and take minimal disk space. If you want to remove them, delete the files in src/main/java/com/threatpipeline/search/, the test file in src/test/java/com/threatpipeline/search/, and revert the additions to docker-compose.yml, pom.xml, and application.properties.

Nice Work!

Nice Work!

You just built a production-grade search and observability layer on top of your Threat Event Pipeline. Security analysts can now hunt for threats with sub-second full-text search, monitor system health in real time, and trust that structured logs will integrate with any aggregator.

Here's what you learned:

  • Indexed threat events from PostgreSQL into Elasticsearch and built a search API with multi_match full-text queries plus filters for severity, source IP, event type, and date range.
  • Wired Micrometer, Prometheus, and Grafana into a live monitoring dashboard showing search latency percentiles, request throughput, error rates, and JVM health.
  • Replaced unstructured text logs with machine-parseable structured JSON using logstash-logback-encoder and proved the entire search pipeline works end-to-end with Testcontainers integration tests against a real Elasticsearch instance.
  • Secret Mission: Added distributed tracing with Zipkin to correlate requests across the entire pipeline with trace IDs in both logs and the Zipkin UI.

Ready to quiz yourself?