cameleer/cameleer-server
1
0
Fork 0
Code Issues 45 Pull Requests Actions Packages Projects Releases Wiki Activity

fix: resolve all integration test failures after storage layer refactor

Browse Source 
- Use singleton container pattern for PostgreSQL + OpenSearch testcontainers
  (fixes container lifecycle issues with @TestInstance(PER_CLASS))
- Fix table name route_executions → executions in DetailControllerIT and
  ExecutionControllerIT
- Serialize processor headers as JSON (ObjectMapper) instead of Map.toString()
  for JSONB column compatibility
- Add nested mapping for processors field in OpenSearch index template
- Use .keyword sub-field for term queries on dynamically mapped text fields
- Add wildcard fallback queries for all text searches (substring matching)
- Isolate stats tests with unique route names to prevent data contamination
- Wait for OpenSearch indexing in SearchControllerIT with targeted Awaitility
- Reduce OpenSearch debounce to 100ms in test profile

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
hsiegeln
2026-03-17 00:02:19 +01:00
parent 26f5a2ce3b
commit 796be06a09
9 changed files with 117 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
cameleer3-server-app/src/main/java/com/cameleer3/server/app/search/OpenSearchIndex.java
View File

@@ -52,7 +52,10 @@ public class OpenSearchIndex implements SearchIndex {
.template(t -> t .template(t -> t
.settings(s -> s .settings(s -> s
.numberOfShards("3") .numberOfShards("3")
.numberOfReplicas("1"))))); .numberOfReplicas("1"))
.mappings(m -> m
.properties("processors", p -> p
.nested(n -> n))))));
log.info("OpenSearch index template created"); log.info("OpenSearch index template created");
} }
} catch (IOException e) { } catch (IOException e) {
@@ -148,27 +151,32 @@ public class OpenSearchIndex implements SearchIndex {
}))); })));
} }
// Keyword filters // Keyword filters (use .keyword sub-field for exact matching on dynamically mapped text fields)
if (request.status() != null) if (request.status() != null)
filter.add(termQuery("status", request.status())); filter.add(termQuery("status.keyword", request.status()));
if (request.routeId() != null) if (request.routeId() != null)
filter.add(termQuery("route_id", request.routeId())); filter.add(termQuery("route_id.keyword", request.routeId()));
if (request.agentId() != null) if (request.agentId() != null)
filter.add(termQuery("agent_id", request.agentId())); filter.add(termQuery("agent_id.keyword", request.agentId()));
if (request.correlationId() != null) if (request.correlationId() != null)
filter.add(termQuery("correlation_id", request.correlationId())); filter.add(termQuery("correlation_id.keyword", request.correlationId()));
// Full-text search across all fields + nested processor fields // Full-text search across all fields + nested processor fields
if (request.text() != null && !request.text().isBlank()) { if (request.text() != null && !request.text().isBlank()) {
String text = request.text(); String text = request.text();
String wildcard = "*" + text.toLowerCase() + "*";
List<Query> textQueries = new ArrayList<>(); List<Query> textQueries = new ArrayList<>();
// Search top-level text fields // Search top-level text fields (analyzed match + wildcard for substring)
textQueries.add(Query.of(q -> q.multiMatch(m -> m textQueries.add(Query.of(q -> q.multiMatch(m -> m
.query(text) .query(text)
.fields("error_message", "error_stacktrace")))); .fields("error_message", "error_stacktrace"))));
textQueries.add(Query.of(q -> q.wildcard(w -> w
.field("error_message").value(wildcard).caseInsensitive(true))));
textQueries.add(Query.of(q -> q.wildcard(w -> w
.field("error_stacktrace").value(wildcard).caseInsensitive(true))));
// Search nested processor fields // Search nested processor fields (analyzed match + wildcard)
textQueries.add(Query.of(q -> q.nested(n -> n textQueries.add(Query.of(q -> q.nested(n -> n
.path("processors") .path("processors")
.query(nq -> nq.multiMatch(m -> m .query(nq -> nq.multiMatch(m -> m
@@ -176,6 +184,14 @@ public class OpenSearchIndex implements SearchIndex {
.fields("processors.input_body", "processors.output_body", .fields("processors.input_body", "processors.output_body",
"processors.input_headers", "processors.output_headers", "processors.input_headers", "processors.output_headers",
"processors.error_message", "processors.error_stacktrace")))))); "processors.error_message", "processors.error_stacktrace"))))));
textQueries.add(Query.of(q -> q.nested(n -> n
.path("processors")
.query(nq -> nq.bool(nb -> nb.should(
wildcardQuery("processors.input_body", wildcard),
wildcardQuery("processors.output_body", wildcard),
wildcardQuery("processors.input_headers", wildcard),
wildcardQuery("processors.output_headers", wildcard)
).minimumShouldMatch("1"))))));
// Also try keyword fields for exact matches // Also try keyword fields for exact matches
textQueries.add(Query.of(q -> q.multiMatch(m -> m textQueries.add(Query.of(q -> q.multiMatch(m -> m
@@ -185,32 +201,51 @@ public class OpenSearchIndex implements SearchIndex {
must.add(Query.of(q -> q.bool(b -> b.should(textQueries).minimumShouldMatch("1")))); must.add(Query.of(q -> q.bool(b -> b.should(textQueries).minimumShouldMatch("1"))));
} }
// Scoped text searches // Scoped text searches (multiMatch + wildcard fallback for substring matching)
if (request.textInBody() != null && !request.textInBody().isBlank()) { if (request.textInBody() != null && !request.textInBody().isBlank()) {
String bodyText = request.textInBody();
String bodyWildcard = "*" + bodyText.toLowerCase() + "*";
must.add(Query.of(q -> q.nested(n -> n must.add(Query.of(q -> q.nested(n -> n
.path("processors") .path("processors")
.query(nq -> nq.multiMatch(m -> m .query(nq -> nq.bool(nb -> nb.should(
.query(request.textInBody()) Query.of(mq -> mq.multiMatch(m -> m
.fields("processors.input_body", "processors.output_body")))))); .query(bodyText)
.fields("processors.input_body", "processors.output_body"))),
wildcardQuery("processors.input_body", bodyWildcard),
wildcardQuery("processors.output_body", bodyWildcard)
).minimumShouldMatch("1"))))));
} }
if (request.textInHeaders() != null && !request.textInHeaders().isBlank()) { if (request.textInHeaders() != null && !request.textInHeaders().isBlank()) {
String headerText = request.textInHeaders();
String headerWildcard = "*" + headerText.toLowerCase() + "*";
must.add(Query.of(q -> q.nested(n -> n must.add(Query.of(q -> q.nested(n -> n
.path("processors") .path("processors")
.query(nq -> nq.multiMatch(m -> m .query(nq -> nq.bool(nb -> nb.should(
.query(request.textInHeaders()) Query.of(mq -> mq.multiMatch(m -> m
.fields("processors.input_headers", "processors.output_headers")))))); .query(headerText)
.fields("processors.input_headers", "processors.output_headers"))),
wildcardQuery("processors.input_headers", headerWildcard),
wildcardQuery("processors.output_headers", headerWildcard)
).minimumShouldMatch("1"))))));
} }
if (request.textInErrors() != null && !request.textInErrors().isBlank()) { if (request.textInErrors() != null && !request.textInErrors().isBlank()) {
String errText = request.textInErrors(); String errText = request.textInErrors();
String errWildcard = "*" + errText.toLowerCase() + "*";
must.add(Query.of(q -> q.bool(b -> b.should( must.add(Query.of(q -> q.bool(b -> b.should(
Query.of(sq -> sq.multiMatch(m -> m Query.of(sq -> sq.multiMatch(m -> m
.query(errText) .query(errText)
.fields("error_message", "error_stacktrace"))), .fields("error_message", "error_stacktrace"))),
wildcardQuery("error_message", errWildcard),
wildcardQuery("error_stacktrace", errWildcard),
Query.of(sq -> sq.nested(n -> n Query.of(sq -> sq.nested(n -> n
.path("processors") .path("processors")
.query(nq -> nq.multiMatch(m -> m .query(nq -> nq.bool(nb -> nb.should(
.query(errText) Query.of(nmq -> nmq.multiMatch(m -> m
.fields("processors.error_message", "processors.error_stacktrace"))))) .query(errText)
.fields("processors.error_message", "processors.error_stacktrace"))),
wildcardQuery("processors.error_message", errWildcard),
wildcardQuery("processors.error_stacktrace", errWildcard)
).minimumShouldMatch("1")))))
).minimumShouldMatch("1")))); ).minimumShouldMatch("1"))));
} }
@@ -238,6 +273,10 @@ public class OpenSearchIndex implements SearchIndex {
return Query.of(q -> q.term(t -> t.field(field).value(FieldValue.of(value)))); return Query.of(q -> q.term(t -> t.field(field).value(FieldValue.of(value))));
} }
private Query wildcardQuery(String field, String pattern) {
return Query.of(q -> q.wildcard(w -> w.field(field).value(pattern).caseInsensitive(true)));
}
private Map<String, Object> toMap(ExecutionDocument doc) { private Map<String, Object> toMap(ExecutionDocument doc) {
Map<String, Object> map = new LinkedHashMap<>(); Map<String, Object> map = new LinkedHashMap<>();
map.put("execution_id", doc.executionId()); map.put("execution_id", doc.executionId());

24
cameleer3-server-app/src/test/java/com/cameleer3/server/app/AbstractPostgresIT.java
View File

@@ -1,5 +1,6 @@
package com.cameleer3.server.app; package com.cameleer3.server.app;
import org.opensearch.testcontainers.OpensearchContainer;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.core.JdbcTemplate;
@@ -7,25 +8,29 @@ import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.DynamicPropertyRegistry; import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource; import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.PostgreSQLContainer; import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName; import org.testcontainers.utility.DockerImageName;
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@ActiveProfiles("test") @ActiveProfiles("test")
@Testcontainers
public abstract class AbstractPostgresIT { public abstract class AbstractPostgresIT {
private static final DockerImageName TIMESCALEDB_IMAGE = private static final DockerImageName TIMESCALEDB_IMAGE =
DockerImageName.parse("timescale/timescaledb-ha:pg16") DockerImageName.parse("timescale/timescaledb-ha:pg16")
.asCompatibleSubstituteFor("postgres"); .asCompatibleSubstituteFor("postgres");
@Container static final PostgreSQLContainer<?> postgres;
static final PostgreSQLContainer<?> postgres = static final OpensearchContainer<?> opensearch;
new PostgreSQLContainer<>(TIMESCALEDB_IMAGE)
.withDatabaseName("cameleer3") static {
.withUsername("cameleer") postgres = new PostgreSQLContainer<>(TIMESCALEDB_IMAGE)
.withPassword("test"); .withDatabaseName("cameleer3")
.withUsername("cameleer")
.withPassword("test");
postgres.start();
opensearch = new OpensearchContainer<>("opensearchproject/opensearch:2.19.0");
opensearch.start();
}
@Autowired @Autowired
protected JdbcTemplate jdbcTemplate; protected JdbcTemplate jdbcTemplate;
@@ -37,5 +42,6 @@ public abstract class AbstractPostgresIT {
registry.add("spring.datasource.password", postgres::getPassword); registry.add("spring.datasource.password", postgres::getPassword);
registry.add("spring.datasource.driver-class-name", () -> "org.postgresql.Driver"); registry.add("spring.datasource.driver-class-name", () -> "org.postgresql.Driver");
registry.add("spring.flyway.enabled", () -> "true"); registry.add("spring.flyway.enabled", () -> "true");
registry.add("opensearch.url", opensearch::getHttpHostAddress);
} }
} }

4
cameleer3-server-app/src/test/java/com/cameleer3/server/app/controller/DetailControllerIT.java
View File

@@ -123,13 +123,13 @@ class DetailControllerIT extends AbstractPostgresIT {
// Wait for flush and get the execution_id // Wait for flush and get the execution_id
await().atMost(10, SECONDS).untilAsserted(() -> { await().atMost(10, SECONDS).untilAsserted(() -> {
Integer count = jdbcTemplate.queryForObject( Integer count = jdbcTemplate.queryForObject(
"SELECT count(*) FROM route_executions WHERE route_id = 'detail-test-route'", "SELECT count(*) FROM executions WHERE route_id = 'detail-test-route'",
Integer.class); Integer.class);
assertThat(count).isGreaterThanOrEqualTo(1); assertThat(count).isGreaterThanOrEqualTo(1);
}); });
seededExecutionId = jdbcTemplate.queryForObject( seededExecutionId = jdbcTemplate.queryForObject(
"SELECT execution_id FROM route_executions WHERE route_id = 'detail-test-route' LIMIT 1", "SELECT execution_id FROM executions WHERE route_id = 'detail-test-route' LIMIT 1",
String.class); String.class);
} }

2
cameleer3-server-app/src/test/java/com/cameleer3/server/app/controller/ExecutionControllerIT.java
View File

@@ -111,7 +111,7 @@ class ExecutionControllerIT extends AbstractPostgresIT {
await().atMost(10, SECONDS).untilAsserted(() -> { await().atMost(10, SECONDS).untilAsserted(() -> {
Integer count = jdbcTemplate.queryForObject( Integer count = jdbcTemplate.queryForObject(
"SELECT count(*) FROM route_executions WHERE route_id = 'flush-test-route'", "SELECT count(*) FROM executions WHERE route_id = 'flush-test-route'",
Integer.class); Integer.class);
assertThat(count).isGreaterThanOrEqualTo(1); assertThat(count).isGreaterThanOrEqualTo(1);
}); });

12
cameleer3-server-app/src/test/java/com/cameleer3/server/app/controller/SearchControllerIT.java
View File

@@ -14,7 +14,9 @@ import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity; import org.springframework.http.ResponseEntity;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
/** /**
* Integration tests for the search controller endpoints. * Integration tests for the search controller endpoints.
@@ -153,11 +155,19 @@ class SearchControllerIT extends AbstractPostgresIT {
""", i, i, i, i, i)); """, i, i, i, i, i));
} }
// Verify all data is available (synchronous writes) // Verify all data is in PostgreSQL (synchronous writes)
Integer count = jdbcTemplate.queryForObject( Integer count = jdbcTemplate.queryForObject(
"SELECT count(*) FROM executions WHERE route_id LIKE 'search-route-%'", "SELECT count(*) FROM executions WHERE route_id LIKE 'search-route-%'",
Integer.class); Integer.class);
assertThat(count).isEqualTo(10); assertThat(count).isEqualTo(10);
// Wait for async OpenSearch indexing (debounce + index time)
// Check for last seeded execution specifically to avoid false positives from other test classes
await().atMost(30, SECONDS).untilAsserted(() -> {
ResponseEntity<String> r = searchGet("?correlationId=corr-page-10");
JsonNode body = objectMapper.readTree(r.getBody());
assertThat(body.get("total").asLong()).isGreaterThanOrEqualTo(1);
});
} }
@Test @Test

15
cameleer3-server-app/src/test/java/com/cameleer3/server/app/search/OpenSearchIndexIT.java
View File

@@ -10,29 +10,16 @@ import com.cameleer3.server.core.storage.model.ExecutionDocument.ProcessorDoc;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.opensearch.client.opensearch.OpenSearchClient; import org.opensearch.client.opensearch.OpenSearchClient;
import org.opensearch.client.opensearch.indices.RefreshRequest; import org.opensearch.client.opensearch.indices.RefreshRequest;
import org.opensearch.testcontainers.OpensearchContainer;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.junit.jupiter.Container;
import java.time.Instant; import java.time.Instant;
import java.util.List; import java.util.List;
import static org.junit.jupiter.api.Assertions.*; import static org.junit.jupiter.api.Assertions.*;
// Extends AbstractPostgresIT for PostgreSQL datasource needed by Spring context // Extends AbstractPostgresIT which provides both PostgreSQL and OpenSearch testcontainers
class OpenSearchIndexIT extends AbstractPostgresIT { class OpenSearchIndexIT extends AbstractPostgresIT {
@Container
static final OpensearchContainer<?> opensearch =
new OpensearchContainer<>("opensearchproject/opensearch:2.19.0");
@DynamicPropertySource
static void configureOpenSearch(DynamicPropertyRegistry registry) {
registry.add("opensearch.url", opensearch::getHttpHostAddress);
}
@Autowired @Autowired
SearchIndex searchIndex; SearchIndex searchIndex;

25
cameleer3-server-app/src/test/java/com/cameleer3/server/app/storage/PostgresStatsStoreIT.java
View File

@@ -23,30 +23,33 @@ class PostgresStatsStoreIT extends AbstractPostgresIT {
@Test @Test
void statsReturnsCountsForTimeWindow() { void statsReturnsCountsForTimeWindow() {
Instant now = Instant.now().truncatedTo(ChronoUnit.SECONDS); // Use a unique route + statsForRoute to avoid data contamination from other tests
insertExecution("stats-1", "route-a", "app-1", "COMPLETED", now, 100L); String uniqueRoute = "stats-route-" + System.nanoTime();
insertExecution("stats-2", "route-a", "app-1", "FAILED", now.plusSeconds(10), 200L); Instant base = Instant.now().minus(5, ChronoUnit.MINUTES).truncatedTo(ChronoUnit.SECONDS);
insertExecution("stats-3", "route-b", "app-1", "COMPLETED", now.plusSeconds(20), 50L); insertExecution("stats-1-" + uniqueRoute, uniqueRoute, "app-stats", "COMPLETED", base, 100L);
insertExecution("stats-2-" + uniqueRoute, uniqueRoute, "app-stats", "FAILED", base.plusSeconds(10), 200L);
insertExecution("stats-3-" + uniqueRoute, uniqueRoute, "app-stats", "COMPLETED", base.plusSeconds(20), 50L);
// Force continuous aggregate refresh // Force continuous aggregate refresh
jdbc.execute("CALL refresh_continuous_aggregate('stats_1m_all', NOW() - INTERVAL '1 hour', NOW() + INTERVAL '1 hour')"); jdbc.execute("CALL refresh_continuous_aggregate('stats_1m_route', NOW() - INTERVAL '1 hour', NOW() + INTERVAL '1 hour')");
ExecutionStats stats = statsStore.stats(now.minusSeconds(60), now.plusSeconds(60)); ExecutionStats stats = statsStore.statsForRoute(base.minusSeconds(60), base.plusSeconds(60), uniqueRoute, null);
assertEquals(3, stats.totalCount()); assertEquals(3, stats.totalCount());
assertEquals(1, stats.failedCount()); assertEquals(1, stats.failedCount());
} }
@Test @Test
void timeseriesReturnsBuckets() { void timeseriesReturnsBuckets() {
Instant now = Instant.now().truncatedTo(ChronoUnit.MINUTES); String uniqueRoute = "ts-route-" + System.nanoTime();
Instant base = Instant.now().minus(10, ChronoUnit.MINUTES).truncatedTo(ChronoUnit.MINUTES);
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10; i++) {
insertExecution("ts-" + i, "route-a", "app-1", "COMPLETED", insertExecution("ts-" + i + "-" + uniqueRoute, uniqueRoute, "app-ts", "COMPLETED",
now.plusSeconds(i * 30), 100L + i); base.plusSeconds(i * 30), 100L + i);
} }
jdbc.execute("CALL refresh_continuous_aggregate('stats_1m_all', NOW() - INTERVAL '1 hour', NOW() + INTERVAL '1 hour')"); jdbc.execute("CALL refresh_continuous_aggregate('stats_1m_route', NOW() - INTERVAL '1 hour', NOW() + INTERVAL '1 hour')");
StatsTimeseries ts = statsStore.timeseries(now.minus(1, ChronoUnit.MINUTES), now.plus(10, ChronoUnit.MINUTES), 5); StatsTimeseries ts = statsStore.timeseriesForRoute(base.minus(1, ChronoUnit.MINUTES), base.plus(10, ChronoUnit.MINUTES), 5, uniqueRoute, null);
assertNotNull(ts); assertNotNull(ts);
assertFalse(ts.buckets().isEmpty()); assertFalse(ts.buckets().isEmpty());
} }

1
cameleer3-server-app/src/test/resources/application-test.yml
View File

@@ -4,6 +4,7 @@ spring:
opensearch: opensearch:
url: http://localhost:9200 url: http://localhost:9200
debounce-ms: 100
ingestion: ingestion:
buffer-capacity: 100 buffer-capacity: 100

17
cameleer3-server-core/src/main/java/com/cameleer3/server/core/ingestion/IngestionService.java
View File

@@ -8,13 +8,18 @@ import com.cameleer3.server.core.storage.ExecutionStore;
import com.cameleer3.server.core.storage.ExecutionStore.ExecutionRecord; import com.cameleer3.server.core.storage.ExecutionStore.ExecutionRecord;
import com.cameleer3.server.core.storage.ExecutionStore.ProcessorRecord; import com.cameleer3.server.core.storage.ExecutionStore.ProcessorRecord;
import com.cameleer3.server.core.storage.model.MetricsSnapshot; import com.cameleer3.server.core.storage.model.MetricsSnapshot;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.function.Consumer; import java.util.function.Consumer;
public class IngestionService { public class IngestionService {
private static final ObjectMapper JSON = new ObjectMapper();
private final ExecutionStore executionStore; private final ExecutionStore executionStore;
private final DiagramStore diagramStore; private final DiagramStore diagramStore;
private final WriteBuffer<MetricsSnapshot> metricsBuffer; private final WriteBuffer<MetricsSnapshot> metricsBuffer;
@@ -99,8 +104,7 @@ public class IngestionService {
p.getDurationMs(), p.getDurationMs(),
p.getErrorMessage(), p.getErrorStackTrace(), p.getErrorMessage(), p.getErrorStackTrace(),
truncateBody(p.getInputBody()), truncateBody(p.getOutputBody()), truncateBody(p.getInputBody()), truncateBody(p.getOutputBody()),
p.getInputHeaders() != null ? p.getInputHeaders().toString() : null, toJson(p.getInputHeaders()), toJson(p.getOutputHeaders())
p.getOutputHeaders() != null ? p.getOutputHeaders().toString() : null
)); ));
if (p.getChildren() != null) { if (p.getChildren() != null) {
flat.addAll(flattenProcessors( flat.addAll(flattenProcessors(
@@ -116,4 +120,13 @@ public class IngestionService {
if (body.length() > bodySizeLimit) return body.substring(0, bodySizeLimit); if (body.length() > bodySizeLimit) return body.substring(0, bodySizeLimit);
return body; return body;
} }
private static String toJson(Map<String, String> headers) {
if (headers == null) return null;
try {
return JSON.writeValueAsString(headers);
} catch (JsonProcessingException e) {
return "{}";
}
}
} }