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

refactor(ingestion): drop dead legacy execution-ingestion path

Browse Source 
ExecutionController was @ConditionalOnMissingBean(ChunkAccumulator.class),
and ChunkAccumulator is registered unconditionally — the legacy controller
never bound in any profile. Even if it had, IngestionService.ingestExecution
called executionStore.upsert(), and the only ExecutionStore impl
(ClickHouseExecutionStore) threw UnsupportedOperationException from upsert
and upsertProcessors. The entire RouteExecution → upsert path was dead code
carrying four transitive dependencies (RouteExecution import, eventPublisher
wiring, body-size-limit config, searchIndexer::onExecutionUpdated hook).

Removed:
- cameleer-server-app/.../controller/ExecutionController.java (whole file)
- ExecutionStore.upsert + upsertProcessors (interface methods)
- ClickHouseExecutionStore.upsert + upsertProcessors (thrower overrides)
- IngestionService.ingestExecution + toExecutionRecord + flattenProcessors
  + hasAnyTraceData + truncateBody + toJson/toJsonObject helpers
- IngestionService constructor now takes (DiagramStore, WriteBuffer<Metrics>);
  dropped ExecutionStore + Consumer<ExecutionUpdatedEvent> + bodySizeLimit
- StorageBeanConfig.ingestionService(...) simplified accordingly

Untouched because still in use:
- ExecutionRecord / ProcessorRecord records (findById / findProcessors /
  SearchIndexer / DetailController)
- SearchIndexer (its onExecutionUpdated never fires now since no-one
  publishes ExecutionUpdatedEvent, but SearchIndexerStats is still
  referenced by ClickHouseAdminController — separate cleanup)
- TaggedExecution record has no remaining callers after this change —
  flagged in core-classes.md as a leftover; separate cleanup.

Rule docs updated:
- .claude/rules/app-classes.md: retired ExecutionController bullet, fixed
  stale URL for ChunkIngestionController (it owns /api/v1/data/executions,
  not /api/v1/ingestion/chunk/executions).
- .claude/rules/core-classes.md: IngestionService surface + note the dead
  TaggedExecution.

Full IT suite post-removal: 560 tests run, 11 F + 1 E — same 12 failures
in the same 3 previously-parked classes (AgentSseControllerIT / SseSigningIT
SSE-timing + ClickHouseStatsStoreIT timezone bug). No regression.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
hsiegeln
2026-04-21 22:50:51 +02:00
parent 56faabcdf1
commit 0f635576a3
7 changed files with 18 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
cameleer-server-app/src/main/java/com/cameleer/server/app/config/StorageBeanConfig.java
View File

@@ -56,13 +56,9 @@ public class StorageBeanConfig {
}
@Bean
public IngestionService ingestionService(ExecutionStore executionStore,
DiagramStore diagramStore,
WriteBuffer<MetricsSnapshot> metricsBuffer,
SearchIndexer searchIndexer,
@Value("${cameleer.server.ingestion.bodysizelimit:16384}") int bodySizeLimit) {
return new IngestionService(executionStore, diagramStore, metricsBuffer,
searchIndexer::onExecutionUpdated, bodySizeLimit);
public IngestionService ingestionService(DiagramStore diagramStore,
WriteBuffer<MetricsSnapshot> metricsBuffer) {
return new IngestionService(diagramStore, metricsBuffer);
}
@Bean

87
cameleer-server-app/src/main/java/com/cameleer/server/app/controller/ExecutionController.java
View File

@@ -1,87 +0,0 @@
package com.cameleer.server.app.controller;
import com.cameleer.common.model.RouteExecution;
import com.cameleer.server.core.agent.AgentInfo;
import com.cameleer.server.core.agent.AgentRegistryService;
import com.cameleer.server.core.ingestion.ChunkAccumulator;
import com.cameleer.server.core.ingestion.IngestionService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.http.ResponseEntity;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
* Legacy ingestion endpoint for route execution data (PostgreSQL path).
* <p>
* Accepts both single {@link RouteExecution} and arrays. Data is written
* synchronously to PostgreSQL via {@link IngestionService}.
* <p>
* Only active when ClickHouse is disabled — when ClickHouse is enabled,
* {@link ChunkIngestionController} takes over the {@code /executions} mapping.
*/
@RestController
@RequestMapping("/api/v1/data")
@ConditionalOnMissingBean(ChunkAccumulator.class)
@Tag(name = "Ingestion", description = "Data ingestion endpoints")
public class ExecutionController {
private final IngestionService ingestionService;
private final AgentRegistryService registryService;
private final ObjectMapper objectMapper;
public ExecutionController(IngestionService ingestionService,
AgentRegistryService registryService,
ObjectMapper objectMapper) {
this.ingestionService = ingestionService;
this.registryService = registryService;
this.objectMapper = objectMapper;
}
@PostMapping("/executions")
@Operation(summary = "Ingest route execution data",
description = "Accepts a single RouteExecution or an array of RouteExecutions")
@ApiResponse(responseCode = "202", description = "Data accepted for processing")
public ResponseEntity<Void> ingestExecutions(@RequestBody String body) throws JsonProcessingException {
String instanceId = extractAgentId();
String applicationId = resolveApplicationId(instanceId);
List<RouteExecution> executions = parsePayload(body);
for (RouteExecution execution : executions) {
ingestionService.ingestExecution(instanceId, applicationId, execution);
}
return ResponseEntity.accepted().build();
}
private String extractAgentId() {
Authentication auth = SecurityContextHolder.getContext().getAuthentication();
return auth != null ? auth.getName() : "";
}
private String resolveApplicationId(String instanceId) {
AgentInfo agent = registryService.findById(instanceId);
return agent != null ? agent.applicationId() : "";
}
private List<RouteExecution> parsePayload(String body) throws JsonProcessingException {
String trimmed = body.strip();
if (trimmed.startsWith("[")) {
return objectMapper.readValue(trimmed, new TypeReference<>() {});
} else {
RouteExecution single = objectMapper.readValue(trimmed, RouteExecution.class);
return List.of(single);
}
}
}

14
cameleer-server-app/src/main/java/com/cameleer/server/app/storage/ClickHouseExecutionStore.java
View File

@@ -282,20 +282,6 @@ public class ClickHouseExecutionStore implements ExecutionStore {
return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0));
}
// --- ExecutionStore interface: write methods (unsupported, use chunked pipeline) ---
@Override
public void upsert(ExecutionRecord execution) {
throw new UnsupportedOperationException("ClickHouse writes use the chunked pipeline");
}
@Override
public void upsertProcessors(String executionId, Instant startTime,
String applicationId, String routeId,
List<ProcessorRecord> processors) {
throw new UnsupportedOperationException("ClickHouse writes use the chunked pipeline");
}
// --- Row mappers ---
private static ExecutionRecord mapExecutionRecord(ResultSet rs) throws SQLException {