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

feat: implement multitenancy with tenant isolation + environment support
All checks were successful
CI / cleanup-branch (push) Has been skipped
Details
CI / build (push) Successful in 1m8s
Details
CI / docker (push) Successful in 42s
Details
CI / deploy-feature (push) Has been skipped
Details
CI / deploy (push) Successful in 1m25s
Details

Browse Source 
Adds configurable tenant ID (CAMELEER_TENANT_ID env var, default:
"default") and environment as a first-class concept. Each server
instance serves one tenant with multiple environments.

Changes across 36 files:
- TenantProperties config bean for tenant ID injection
- AgentInfo: added environmentId field
- AgentRegistrationRequest: added environmentId field
- All 9 ClickHouse stores: inject tenant ID, replace hardcoded
  "default" constant, add environment to writes/reads
- ChunkAccumulator: configurable tenant ID + environment resolver
- MergedExecution/ProcessorBatch/BufferedLogEntry: added environment
- ClickHouse init.sql: added environment column to all tables,
  updated ORDER BY (tenant→time→env→app), added tenant_id to
  usage_events, updated all MV GROUP BY clauses
- Controllers: pass environmentId through registration/auto-heal
- K8s deploy: added CAMELEER_TENANT_ID env var
- All tests updated for new signatures

Closes #123

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
hsiegeln
2026-04-04 15:00:18 +02:00
parent ee7226cf1c
commit a188308ec5
36 changed files with 310 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
cameleer3-server-app/src/main/java/com/cameleer3/server/app/config/StorageBeanConfig.java
View File

@@ -10,6 +10,8 @@ import com.cameleer3.server.app.storage.ClickHouseStatsStore;
import com.cameleer3.server.core.admin.AuditRepository;
import com.cameleer3.server.core.admin.AuditService;
import com.cameleer3.server.core.agent.AgentEventRepository;
import com.cameleer3.server.core.agent.AgentInfo;
import com.cameleer3.server.core.agent.AgentRegistryService;
import com.cameleer3.server.core.detail.DetailService;
import com.cameleer3.server.core.indexing.SearchIndexer;
import com.cameleer3.server.app.ingestion.ExecutionFlushScheduler;
@@ -63,34 +65,45 @@ public class StorageBeanConfig {
@Bean
public MetricsStore clickHouseMetricsStore(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseMetricsStore(clickHouseJdbc);
return new ClickHouseMetricsStore(tenantProperties.getId(), clickHouseJdbc);
}
@Bean
public MetricsQueryStore clickHouseMetricsQueryStore(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseMetricsQueryStore(clickHouseJdbc);
return new ClickHouseMetricsQueryStore(tenantProperties.getId(), clickHouseJdbc);
}
// ── Execution Store ──────────────────────────────────────────────────
@Bean
public ClickHouseExecutionStore clickHouseExecutionStore(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseExecutionStore(clickHouseJdbc);
return new ClickHouseExecutionStore(tenantProperties.getId(), clickHouseJdbc);
}
@Bean
public ChunkAccumulator chunkAccumulator(
TenantProperties tenantProperties,
WriteBuffer<MergedExecution> executionBuffer,
WriteBuffer<ChunkAccumulator.ProcessorBatch> processorBatchBuffer,
DiagramStore diagramStore) {
DiagramStore diagramStore,
AgentRegistryService registryService) {
return new ChunkAccumulator(
tenantProperties.getId(),
executionBuffer::offerOrWarn,
processorBatchBuffer::offerOrWarn,
diagramStore,
java.time.Duration.ofMinutes(5));
java.time.Duration.ofMinutes(5),
instanceId -> {
AgentInfo agent = registryService.findById(instanceId);
return agent != null && agent.environmentId() != null
? agent.environmentId() : "default";
});
}
@Bean
@@ -108,40 +121,45 @@ public class StorageBeanConfig {
@Bean
public SearchIndex clickHouseSearchIndex(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseSearchIndex(clickHouseJdbc);
return new ClickHouseSearchIndex(tenantProperties.getId(), clickHouseJdbc);
}
// ── ClickHouse Stats Store ─────────────────────────────────────────
@Bean
public StatsStore clickHouseStatsStore(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseStatsStore(clickHouseJdbc);
return new ClickHouseStatsStore(tenantProperties.getId(), clickHouseJdbc);
}
// ── ClickHouse Diagram Store ──────────────────────────────────────
@Bean
public DiagramStore clickHouseDiagramStore(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseDiagramStore(clickHouseJdbc);
return new ClickHouseDiagramStore(tenantProperties.getId(), clickHouseJdbc);
}
// ── ClickHouse Agent Event Repository ─────────────────────────────
@Bean
public AgentEventRepository clickHouseAgentEventRepository(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseAgentEventRepository(clickHouseJdbc);
return new ClickHouseAgentEventRepository(tenantProperties.getId(), clickHouseJdbc);
}
// ── ClickHouse Log Store ──────────────────────────────────────────
@Bean
public ClickHouseLogStore clickHouseLogStore(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseLogStore(clickHouseJdbc);
return new ClickHouseLogStore(tenantProperties.getId(), clickHouseJdbc);
}
// ── Usage Analytics ──────────────────────────────────────────────
@@ -149,8 +167,9 @@ public class StorageBeanConfig {
@Bean
@ConditionalOnProperty(name = "clickhouse.enabled", havingValue = "true")
public ClickHouseUsageTracker clickHouseUsageTracker(
TenantProperties tenantProperties,
@Qualifier("clickHouseJdbcTemplate") JdbcTemplate clickHouseJdbc) {
return new ClickHouseUsageTracker(clickHouseJdbc,
return new ClickHouseUsageTracker(tenantProperties.getId(), clickHouseJdbc,
new com.cameleer3.server.core.ingestion.WriteBuffer<>(5000));
}

19
cameleer3-server-app/src/main/java/com/cameleer3/server/app/config/TenantProperties.java Normal file
View File

@@ -0,0 +1,19 @@
package com.cameleer3.server.app.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
@Component
@ConfigurationProperties(prefix = "cameleer.tenant")
public class TenantProperties {
private String id = "default";
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
}

6
cameleer3-server-app/src/main/java/com/cameleer3/server/app/controller/AgentRegistrationController.java
View File

@@ -115,11 +115,13 @@ public class AgentRegistrationController {
}
String application = request.applicationId() != null ? request.applicationId() : "default";
String environmentId = request.environmentId() != null ? request.environmentId() : "default";
List<String> routeIds = request.routeIds() != null ? request.routeIds() : List.of();
var capabilities = request.capabilities() != null ? request.capabilities() : Collections.<String, Object>emptyMap();
AgentInfo agent = registryService.register(
request.instanceId(), request.displayName(), application, request.version(), routeIds, capabilities);
request.instanceId(), request.displayName(), application, environmentId,
request.version(), routeIds, capabilities);
log.info("Agent registered: {} (name={}, application={})", request.instanceId(), request.displayName(), application);
agentEventService.recordEvent(request.instanceId(), application, "REGISTERED",
@@ -210,7 +212,7 @@ public class AgentRegistrationController {
if (jwtResult != null) {
String application = jwtResult.application() != null ? jwtResult.application() : "default";
Map<String, Object> caps = capabilities != null ? capabilities : Map.of();
registryService.register(id, id, application, "unknown",
registryService.register(id, id, application, "default", "unknown",
List.of(), caps);
registryService.heartbeat(id);
log.info("Auto-registered agent {} (app={}) from heartbeat after server restart", id, application);

2
cameleer3-server-app/src/main/java/com/cameleer3/server/app/controller/AgentSseController.java
View File

@@ -67,7 +67,7 @@ public class AgentSseController {
JwtAuthenticationFilter.JWT_RESULT_ATTR);
if (jwtResult != null) {
String application = jwtResult.application() != null ? jwtResult.application() : "default";
registryService.register(id, id, application, "unknown", List.of(), Map.of());
registryService.register(id, id, application, "default", "unknown", List.of(), Map.of());
log.info("Auto-registered agent {} (app={}) from SSE connect after server restart", id, application);
} else {
throw new ResponseStatusException(HttpStatus.NOT_FOUND, "Agent not found: " + id);

15
cameleer3-server-app/src/main/java/com/cameleer3/server/app/controller/LogIngestionController.java
View File

@@ -8,6 +8,7 @@ import com.cameleer3.server.core.agent.AgentRegistryService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
import io.swagger.v3.oas.annotations.tags.Tag;
import com.cameleer3.server.app.config.TenantProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
@@ -27,11 +28,14 @@ public class LogIngestionController {
private final WriteBuffer<BufferedLogEntry> logBuffer;
private final AgentRegistryService registryService;
private final TenantProperties tenantProperties;
public LogIngestionController(WriteBuffer<BufferedLogEntry> logBuffer,
AgentRegistryService registryService) {
AgentRegistryService registryService,
TenantProperties tenantProperties) {
this.logBuffer = logBuffer;
this.registryService = registryService;
this.tenantProperties = tenantProperties;
}
@PostMapping("/logs")
@@ -44,8 +48,10 @@ public class LogIngestionController {
if (batch.getEntries() != null && !batch.getEntries().isEmpty()) {
log.debug("Received {} log entries from instance={}, app={}", batch.getEntries().size(), instanceId, applicationId);
String environment = resolveEnvironment(instanceId);
for (var entry : batch.getEntries()) {
logBuffer.offerOrWarn(new BufferedLogEntry(instanceId, applicationId, entry));
logBuffer.offerOrWarn(new BufferedLogEntry(
tenantProperties.getId(), environment, instanceId, applicationId, entry));
}
}
@@ -61,4 +67,9 @@ public class LogIngestionController {
AgentInfo agent = registryService.findById(instanceId);
return agent != null ? agent.applicationId() : "";
}
private String resolveEnvironment(String instanceId) {
AgentInfo agent = registryService.findById(instanceId);
return agent != null && agent.environmentId() != null ? agent.environmentId() : "default";
}
}

1
cameleer3-server-app/src/main/java/com/cameleer3/server/app/dto/AgentRegistrationRequest.java
View File

@@ -11,6 +11,7 @@ public record AgentRegistrationRequest(
@NotNull String instanceId,
@NotNull String displayName,
@Schema(defaultValue = "default") String applicationId,
@Schema(defaultValue = "default") String environmentId,
String version,
List<String> routeIds,
Map<String, Object> capabilities

57
cameleer3-server-app/src/main/java/com/cameleer3/server/app/search/ClickHouseLogStore.java
View File

@@ -29,12 +29,13 @@ import java.util.Map;
public class ClickHouseLogStore implements LogIndex {
private static final Logger log = LoggerFactory.getLogger(ClickHouseLogStore.class);
private static final String TENANT = "default";
private static final DateTimeFormatter ISO_FMT = DateTimeFormatter.ISO_INSTANT;
private final String tenantId;
private final JdbcTemplate jdbc;
public ClickHouseLogStore(JdbcTemplate jdbc) {
public ClickHouseLogStore(String tenantId, JdbcTemplate jdbc) {
this.tenantId = tenantId;
this.jdbc = jdbc;
}
@@ -46,23 +47,24 @@ public class ClickHouseLogStore implements LogIndex {
String sql = "INSERT INTO logs (tenant_id, timestamp, application, instance_id, level, " +
"logger_name, message, thread_name, stack_trace, exchange_id, mdc) " +
"VALUES ('default', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
jdbc.batchUpdate(sql, entries, entries.size(), (ps, entry) -> {
Instant ts = entry.getTimestamp() != null ? entry.getTimestamp() : Instant.now();
ps.setTimestamp(1, Timestamp.from(ts));
ps.setString(2, applicationId);
ps.setString(3, instanceId);
ps.setString(4, entry.getLevel() != null ? entry.getLevel() : "");
ps.setString(5, entry.getLoggerName() != null ? entry.getLoggerName() : "");
ps.setString(6, entry.getMessage() != null ? entry.getMessage() : "");
ps.setString(7, entry.getThreadName() != null ? entry.getThreadName() : "");
ps.setString(8, entry.getStackTrace() != null ? entry.getStackTrace() : "");
ps.setString(1, tenantId);
ps.setTimestamp(2, Timestamp.from(ts));
ps.setString(3, applicationId);
ps.setString(4, instanceId);
ps.setString(5, entry.getLevel() != null ? entry.getLevel() : "");
ps.setString(6, entry.getLoggerName() != null ? entry.getLoggerName() : "");
ps.setString(7, entry.getMessage() != null ? entry.getMessage() : "");
ps.setString(8, entry.getThreadName() != null ? entry.getThreadName() : "");
ps.setString(9, entry.getStackTrace() != null ? entry.getStackTrace() : "");
Map<String, String> mdc = entry.getMdc() != null ? entry.getMdc() : Collections.emptyMap();
String exchangeId = mdc.getOrDefault("camel.exchangeId", "");
ps.setString(9, exchangeId);
ps.setObject(10, mdc);
ps.setString(10, exchangeId);
ps.setObject(11, mdc);
});
log.debug("Indexed {} log entries for instance={}, app={}", entries.size(), instanceId, applicationId);
@@ -71,26 +73,28 @@ public class ClickHouseLogStore implements LogIndex {
public void insertBufferedBatch(List<BufferedLogEntry> entries) {
if (entries.isEmpty()) return;
String sql = "INSERT INTO logs (tenant_id, timestamp, application, instance_id, level, " +
String sql = "INSERT INTO logs (tenant_id, environment, timestamp, application, instance_id, level, " +
"logger_name, message, thread_name, stack_trace, exchange_id, mdc) " +
"VALUES ('default', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
jdbc.batchUpdate(sql, entries, entries.size(), (ps, ble) -> {
LogEntry entry = ble.entry();
Instant ts = entry.getTimestamp() != null ? entry.getTimestamp() : Instant.now();
ps.setTimestamp(1, Timestamp.from(ts));
ps.setString(2, ble.applicationId());
ps.setString(3, ble.instanceId());
ps.setString(4, entry.getLevel() != null ? entry.getLevel() : "");
ps.setString(5, entry.getLoggerName() != null ? entry.getLoggerName() : "");
ps.setString(6, entry.getMessage() != null ? entry.getMessage() : "");
ps.setString(7, entry.getThreadName() != null ? entry.getThreadName() : "");
ps.setString(8, entry.getStackTrace() != null ? entry.getStackTrace() : "");
ps.setString(1, ble.tenantId() != null ? ble.tenantId() : tenantId);
ps.setString(2, ble.environment() != null ? ble.environment() : "default");
ps.setTimestamp(3, Timestamp.from(ts));
ps.setString(4, ble.applicationId());
ps.setString(5, ble.instanceId());
ps.setString(6, entry.getLevel() != null ? entry.getLevel() : "");
ps.setString(7, entry.getLoggerName() != null ? entry.getLoggerName() : "");
ps.setString(8, entry.getMessage() != null ? entry.getMessage() : "");
ps.setString(9, entry.getThreadName() != null ? entry.getThreadName() : "");
ps.setString(10, entry.getStackTrace() != null ? entry.getStackTrace() : "");
Map<String, String> mdc = entry.getMdc() != null ? entry.getMdc() : Collections.emptyMap();
String exchangeId = mdc.getOrDefault("camel.exchangeId", "");
ps.setString(9, exchangeId);
ps.setObject(10, mdc);
ps.setString(11, exchangeId);
ps.setObject(12, mdc);
});
log.debug("Flushed {} buffered log entries to ClickHouse", entries.size());
@@ -101,7 +105,8 @@ public class ClickHouseLogStore implements LogIndex {
// Build shared WHERE conditions (used by both data and count queries)
List<String> baseConditions = new ArrayList<>();
List<Object> baseParams = new ArrayList<>();
baseConditions.add("tenant_id = 'default'");
baseConditions.add("tenant_id = ?");
baseParams.add(tenantId);
if (request.application() != null && !request.application().isEmpty()) {
baseConditions.add("application = ?");

23
cameleer3-server-app/src/main/java/com/cameleer3/server/app/search/ClickHouseSearchIndex.java
View File

@@ -44,9 +44,11 @@ public class ClickHouseSearchIndex implements SearchIndex {
"applicationId", "application_id"
);
private final String tenantId;
private final JdbcTemplate jdbc;
public ClickHouseSearchIndex(JdbcTemplate jdbc) {
public ClickHouseSearchIndex(String tenantId, JdbcTemplate jdbc) {
this.tenantId = tenantId;
this.jdbc = jdbc;
}
@@ -118,7 +120,8 @@ public class ClickHouseSearchIndex implements SearchIndex {
private String buildWhereClause(SearchRequest request, List<Object> params) {
List<String> conditions = new ArrayList<>();
conditions.add("tenant_id = 'default'");
conditions.add("tenant_id = ?");
params.add(tenantId);
if (request.timeFrom() != null) {
conditions.add("start_time >= ?");
@@ -186,11 +189,12 @@ public class ClickHouseSearchIndex implements SearchIndex {
conditions.add("(execution_id = ? OR correlation_id = ? OR exchange_id = ?"
+ " OR _search_text LIKE ? OR execution_id IN ("
+ "SELECT DISTINCT execution_id FROM processor_executions "
+ "WHERE tenant_id = 'default' AND _search_text LIKE ?))");
+ "WHERE tenant_id = ? AND _search_text LIKE ?))");
params.add(term);
params.add(term);
params.add(term);
params.add(likeTerm);
params.add(tenantId);
params.add(likeTerm);
}
@@ -199,7 +203,8 @@ public class ClickHouseSearchIndex implements SearchIndex {
String likeTerm = "%" + escapeLike(request.textInBody()) + "%";
conditions.add("execution_id IN ("
+ "SELECT DISTINCT execution_id FROM processor_executions "
+ "WHERE tenant_id = 'default' AND (input_body LIKE ? OR output_body LIKE ?))");
+ "WHERE tenant_id = ? AND (input_body LIKE ? OR output_body LIKE ?))");
params.add(tenantId);
params.add(likeTerm);
params.add(likeTerm);
}
@@ -209,7 +214,8 @@ public class ClickHouseSearchIndex implements SearchIndex {
String likeTerm = "%" + escapeLike(request.textInHeaders()) + "%";
conditions.add("execution_id IN ("
+ "SELECT DISTINCT execution_id FROM processor_executions "
+ "WHERE tenant_id = 'default' AND (input_headers LIKE ? OR output_headers LIKE ?))");
+ "WHERE tenant_id = ? AND (input_headers LIKE ? OR output_headers LIKE ?))");
params.add(tenantId);
params.add(likeTerm);
params.add(likeTerm);
}
@@ -219,9 +225,10 @@ public class ClickHouseSearchIndex implements SearchIndex {
String likeTerm = "%" + escapeLike(request.textInErrors()) + "%";
conditions.add("(error_message LIKE ? OR error_stacktrace LIKE ? OR execution_id IN ("
+ "SELECT DISTINCT execution_id FROM processor_executions "
+ "WHERE tenant_id = 'default' AND (error_message LIKE ? OR error_stacktrace LIKE ?)))");
+ "WHERE tenant_id = ? AND (error_message LIKE ? OR error_stacktrace LIKE ?)))");
params.add(likeTerm);
params.add(likeTerm);
params.add(tenantId);
params.add(likeTerm);
params.add(likeTerm);
}
@@ -311,9 +318,9 @@ public class ClickHouseSearchIndex implements SearchIndex {
return jdbc.queryForList("""
SELECT DISTINCT arrayJoin(JSONExtractKeys(attributes)) AS attr_key
FROM executions FINAL
WHERE tenant_id = 'default' AND attributes != '' AND attributes != '{}'
WHERE tenant_id = ? AND attributes != '' AND attributes != '{}'
ORDER BY attr_key
""", String.class);
""", String.class, tenantId);
} catch (Exception e) {
log.error("Failed to query distinct attribute keys", e);
return List.of();

10
cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseAgentEventRepository.java
View File

@@ -17,30 +17,30 @@ import java.util.List;
*/
public class ClickHouseAgentEventRepository implements AgentEventRepository {
private static final String TENANT = "default";
private static final String INSERT_SQL =
"INSERT INTO agent_events (tenant_id, instance_id, application_id, event_type, detail) VALUES (?, ?, ?, ?, ?)";
private static final String SELECT_BASE =
"SELECT 0 AS id, instance_id, application_id, event_type, detail, timestamp FROM agent_events WHERE tenant_id = ?";
private final String tenantId;
private final JdbcTemplate jdbc;
public ClickHouseAgentEventRepository(JdbcTemplate jdbc) {
public ClickHouseAgentEventRepository(String tenantId, JdbcTemplate jdbc) {
this.tenantId = tenantId;
this.jdbc = jdbc;
}
@Override
public void insert(String instanceId, String applicationId, String eventType, String detail) {
jdbc.update(INSERT_SQL, TENANT, instanceId, applicationId, eventType, detail);
jdbc.update(INSERT_SQL, tenantId, instanceId, applicationId, eventType, detail);
}
@Override
public List<AgentEventRecord> query(String applicationId, String instanceId, Instant from, Instant to, int limit) {
var sql = new StringBuilder(SELECT_BASE);
var params = new ArrayList<Object>();
params.add(TENANT);
params.add(tenantId);
if (applicationId != null) {
sql.append(" AND application_id = ?");

18
cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseDiagramStore.java
View File

@@ -39,8 +39,6 @@ public class ClickHouseDiagramStore implements DiagramStore {
private static final Logger log = LoggerFactory.getLogger(ClickHouseDiagramStore.class);
private static final String TENANT = "default";
private static final String INSERT_SQL = """
INSERT INTO route_diagrams
(tenant_id, content_hash, route_id, instance_id, application_id, definition, created_at)
@@ -64,6 +62,7 @@ public class ClickHouseDiagramStore implements DiagramStore {
WHERE tenant_id = ? AND application_id = ?
""";
private final String tenantId;
private final JdbcTemplate jdbc;
private final ObjectMapper objectMapper;
@@ -72,7 +71,8 @@ public class ClickHouseDiagramStore implements DiagramStore {
// contentHash → deserialized RouteGraph
private final ConcurrentHashMap<String, RouteGraph> graphCache = new ConcurrentHashMap<>();
public ClickHouseDiagramStore(JdbcTemplate jdbc) {
public ClickHouseDiagramStore(String tenantId, JdbcTemplate jdbc) {
this.tenantId = tenantId;
this.jdbc = jdbc;
this.objectMapper = new ObjectMapper();
this.objectMapper.registerModule(new JavaTimeModule());
@@ -87,7 +87,7 @@ public class ClickHouseDiagramStore implements DiagramStore {
String key = rs.getString("route_id") + "\0" + rs.getString("instance_id");
hashCache.put(key, rs.getString("content_hash"));
},
TENANT);
tenantId);
log.info("Diagram hash cache warmed: {} entries", hashCache.size());
} catch (Exception e) {
log.warn("Failed to warm diagram hash cache — lookups will fall back to ClickHouse: {}", e.getMessage());
@@ -109,7 +109,7 @@ public class ClickHouseDiagramStore implements DiagramStore {
String routeId = graph.getRouteId() != null ? graph.getRouteId() : "";
jdbc.update(INSERT_SQL,
TENANT,
tenantId,
contentHash,
routeId,
agentId,
@@ -134,7 +134,7 @@ public class ClickHouseDiagramStore implements DiagramStore {
return Optional.of(cached);
}
List<Map<String, Object>> rows = jdbc.queryForList(SELECT_BY_HASH, TENANT, contentHash);
List<Map<String, Object>> rows = jdbc.queryForList(SELECT_BY_HASH, tenantId, contentHash);
if (rows.isEmpty()) {
return Optional.empty();
}
@@ -157,7 +157,7 @@ public class ClickHouseDiagramStore implements DiagramStore {
}
List<Map<String, Object>> rows = jdbc.queryForList(
SELECT_HASH_FOR_ROUTE, TENANT, routeId, agentId);
SELECT_HASH_FOR_ROUTE, tenantId, routeId, agentId);
if (rows.isEmpty()) {
return Optional.empty();
}
@@ -186,7 +186,7 @@ public class ClickHouseDiagramStore implements DiagramStore {
"WHERE tenant_id = ? AND route_id = ? AND instance_id IN (" + placeholders + ") " +
"ORDER BY created_at DESC LIMIT 1";
var params = new ArrayList<Object>();
params.add(TENANT);
params.add(tenantId);
params.add(routeId);
params.addAll(agentIds);
List<Map<String, Object>> rows = jdbc.queryForList(sql, params.toArray());
@@ -200,7 +200,7 @@ public class ClickHouseDiagramStore implements DiagramStore {
public Map<String, String> findProcessorRouteMapping(String applicationId) {
Map<String, String> mapping = new HashMap<>();
List<Map<String, Object>> rows = jdbc.queryForList(
SELECT_DEFINITIONS_FOR_APP, TENANT, applicationId);
SELECT_DEFINITIONS_FOR_APP, tenantId, applicationId);
for (Map<String, Object> row : rows) {
String routeId = (String) row.get("route_id");
String json = (String) row.get("definition");

29
cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseExecutionStore.java
View File

@@ -18,14 +18,16 @@ import java.util.Optional;
public class ClickHouseExecutionStore implements ExecutionStore {
private final String tenantId;
private final JdbcTemplate jdbc;
private final ObjectMapper objectMapper;
public ClickHouseExecutionStore(JdbcTemplate jdbc) {
this(jdbc, new ObjectMapper());
public ClickHouseExecutionStore(String tenantId, JdbcTemplate jdbc) {
this(tenantId, jdbc, new ObjectMapper());
}
public ClickHouseExecutionStore(JdbcTemplate jdbc, ObjectMapper objectMapper) {
public ClickHouseExecutionStore(String tenantId, JdbcTemplate jdbc, ObjectMapper objectMapper) {
this.tenantId = tenantId;
this.jdbc = jdbc;
this.objectMapper = objectMapper;
}
@@ -36,14 +38,14 @@ public class ClickHouseExecutionStore implements ExecutionStore {
jdbc.batchUpdate("""
INSERT INTO executions (
tenant_id, _version, execution_id, route_id, instance_id, application_id,
status, correlation_id, exchange_id, start_time, end_time, duration_ms,
environment, status, correlation_id, exchange_id, start_time, end_time, duration_ms,
error_message, error_stacktrace, error_type, error_category,
root_cause_type, root_cause_message, diagram_content_hash, engine_level,
input_body, output_body, input_headers, output_headers, attributes,
trace_id, span_id, has_trace_data, is_replay,
original_exchange_id, replay_exchange_id
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
executions.stream().map(e -> new Object[]{
nullToEmpty(e.tenantId()),
@@ -52,6 +54,7 @@ public class ClickHouseExecutionStore implements ExecutionStore {
nullToEmpty(e.routeId()),
nullToEmpty(e.instanceId()),
nullToEmpty(e.applicationId()),
nullToEmpty(e.environment()),
nullToEmpty(e.status()),
nullToEmpty(e.correlationId()),
nullToEmpty(e.exchangeId()),
@@ -199,11 +202,11 @@ public class ClickHouseExecutionStore implements ExecutionStore {
error_type, error_category, root_cause_type, root_cause_message,
trace_id, span_id, has_trace_data, is_replay
FROM executions FINAL
WHERE tenant_id = 'default' AND execution_id = ?
WHERE tenant_id = ? AND execution_id = ?
LIMIT 1
""",
(rs, rowNum) -> mapExecutionRecord(rs),
executionId);
tenantId, executionId);
return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0));
}
@@ -219,11 +222,11 @@ public class ClickHouseExecutionStore implements ExecutionStore {
resolved_endpoint_uri, circuit_breaker_state,
fallback_triggered, filter_matched, duplicate_message
FROM processor_executions
WHERE tenant_id = 'default' AND execution_id = ?
WHERE tenant_id = ? AND execution_id = ?
ORDER BY seq
""",
(rs, rowNum) -> mapProcessorRecord(rs),
executionId);
tenantId, executionId);
}
@Override
@@ -238,11 +241,11 @@ public class ClickHouseExecutionStore implements ExecutionStore {
resolved_endpoint_uri, circuit_breaker_state,
fallback_triggered, filter_matched, duplicate_message
FROM processor_executions
WHERE tenant_id = 'default' AND execution_id = ? AND processor_id = ?
WHERE tenant_id = ? AND execution_id = ? AND processor_id = ?
LIMIT 1
""",
(rs, rowNum) -> mapProcessorRecord(rs),
executionId, processorId);
tenantId, executionId, processorId);
return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0));
}
@@ -258,11 +261,11 @@ public class ClickHouseExecutionStore implements ExecutionStore {
resolved_endpoint_uri, circuit_breaker_state,
fallback_triggered, filter_matched, duplicate_message
FROM processor_executions
WHERE tenant_id = 'default' AND execution_id = ? AND seq = ?
WHERE tenant_id = ? AND execution_id = ? AND seq = ?
LIMIT 1
""",
(rs, rowNum) -> mapProcessorRecord(rs),
executionId, seq);
tenantId, executionId, seq);
return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0));
}

8
cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseMetricsQueryStore.java
View File

@@ -13,9 +13,11 @@ import java.util.Map;
public class ClickHouseMetricsQueryStore implements MetricsQueryStore {
private final String tenantId;
private final JdbcTemplate jdbc;
public ClickHouseMetricsQueryStore(JdbcTemplate jdbc) {
public ClickHouseMetricsQueryStore(String tenantId, JdbcTemplate jdbc) {
this.tenantId = tenantId;
this.jdbc = jdbc;
}
@@ -41,7 +43,8 @@ public class ClickHouseMetricsQueryStore implements MetricsQueryStore {
metric_name,
avg(metric_value) AS avg_value
FROM agent_metrics
WHERE instance_id = ?
WHERE tenant_id = ?
AND instance_id = ?
AND collected_at >= ?
AND collected_at < ?
AND metric_name IN (%s)
@@ -50,6 +53,7 @@ public class ClickHouseMetricsQueryStore implements MetricsQueryStore {
""".formatted(intervalSeconds, placeholders);
List<Object> params = new ArrayList<>();
params.add(tenantId);
params.add(instanceId);
params.add(java.sql.Timestamp.from(from));
params.add(java.sql.Timestamp.from(to));

9
cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseMetricsStore.java
View File

@@ -11,9 +11,11 @@ import java.util.Map;
public class ClickHouseMetricsStore implements MetricsStore {
private final String tenantId;
private final JdbcTemplate jdbc;
public ClickHouseMetricsStore(JdbcTemplate jdbc) {
public ClickHouseMetricsStore(String tenantId, JdbcTemplate jdbc) {
this.tenantId = tenantId;
this.jdbc = jdbc;
}
@@ -22,10 +24,11 @@ public class ClickHouseMetricsStore implements MetricsStore {
if (snapshots.isEmpty()) return;
jdbc.batchUpdate("""
INSERT INTO agent_metrics (instance_id, metric_name, metric_value, tags, collected_at)
VALUES (?, ?, ?, ?, ?)
INSERT INTO agent_metrics (tenant_id, instance_id, metric_name, metric_value, tags, collected_at)
VALUES (?, ?, ?, ?, ?, ?)
""",
snapshots.stream().map(s -> new Object[]{
tenantId,
s.instanceId(),
s.metricName(),
s.metricValue(),

34
cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseStatsStore.java
View File

@@ -31,11 +31,11 @@ import java.util.Map;
*/
public class ClickHouseStatsStore implements StatsStore {
private static final String TENANT = "default";
private final String tenantId;
private final JdbcTemplate jdbc;
public ClickHouseStatsStore(JdbcTemplate jdbc) {
public ClickHouseStatsStore(String tenantId, JdbcTemplate jdbc) {
this.tenantId = tenantId;
this.jdbc = jdbc;
}
@@ -117,7 +117,7 @@ public class ClickHouseStatsStore implements StatsStore {
List<Object> params = new ArrayList<>();
params.add(thresholdMs);
params.add(TENANT);
params.add(tenantId);
params.add(Timestamp.from(from));
params.add(Timestamp.from(to));
if (applicationId != null) {
@@ -149,7 +149,7 @@ public class ClickHouseStatsStore implements StatsStore {
jdbc.query(sql, (rs) -> {
result.put(rs.getString("application_id"),
new long[]{rs.getLong("compliant"), rs.getLong("total")});
}, defaultThresholdMs, TENANT, Timestamp.from(from), Timestamp.from(to));
}, defaultThresholdMs, tenantId, Timestamp.from(from), Timestamp.from(to));
return result;
}
@@ -167,7 +167,7 @@ public class ClickHouseStatsStore implements StatsStore {
jdbc.query(sql, (rs) -> {
result.put(rs.getString("route_id"),
new long[]{rs.getLong("compliant"), rs.getLong("total")});
}, thresholdMs, TENANT, Timestamp.from(from), Timestamp.from(to), applicationId);
}, thresholdMs, tenantId, Timestamp.from(from), Timestamp.from(to), applicationId);
return result;
}
@@ -223,13 +223,13 @@ public class ClickHouseStatsStore implements StatsStore {
"ORDER BY c.cnt DESC";
List<Object> fullParams = new ArrayList<>();
fullParams.add(TENANT);
fullParams.add(tenantId);
fullParams.addAll(params);
fullParams.add(limit);
fullParams.add(Timestamp.from(fiveMinAgo));
fullParams.add(Timestamp.from(tenMinAgo));
fullParams.add(Timestamp.from(fiveMinAgo));
fullParams.add(TENANT);
fullParams.add(tenantId);
fullParams.addAll(params);
return jdbc.query(sql, (rs, rowNum) -> {
@@ -253,7 +253,7 @@ public class ClickHouseStatsStore implements StatsStore {
"WHERE tenant_id = ? AND status = 'FAILED' AND start_time >= ? AND start_time < ?";
List<Object> params = new ArrayList<>();
params.add(TENANT);
params.add(tenantId);
params.add(Timestamp.from(from));
params.add(Timestamp.from(to));
if (applicationId != null) {
@@ -275,7 +275,7 @@ public class ClickHouseStatsStore implements StatsStore {
"countMerge(total_count) AS total_count, " +
"countIfMerge(failed_count) AS failed_count " +
"FROM " + view +
" WHERE tenant_id = " + lit(TENANT) +
" WHERE tenant_id = " + lit(tenantId) +
" AND bucket >= " + lit(from) +
" AND bucket < " + lit(to);
if (applicationId != null) {
@@ -327,7 +327,7 @@ public class ClickHouseStatsStore implements StatsStore {
"quantileMerge(0.99)(p99_duration) AS p99_duration, " +
runningCol + " AS active_count " +
"FROM " + view +
" WHERE tenant_id = " + lit(TENANT) +
" WHERE tenant_id = " + lit(tenantId) +
" AND bucket >= " + lit(rangeFrom) +
" AND bucket < " + lit(rangeTo);
for (Filter f : filters) {
@@ -413,7 +413,7 @@ public class ClickHouseStatsStore implements StatsStore {
"quantileMerge(0.99)(p99_duration) AS p99_duration, " +
runningCol + " AS active_count " +
"FROM " + view +
" WHERE tenant_id = " + lit(TENANT) +
" WHERE tenant_id = " + lit(tenantId) +
" AND bucket >= " + lit(from) +
" AND bucket < " + lit(to);
for (Filter f : filters) {
@@ -453,7 +453,7 @@ public class ClickHouseStatsStore implements StatsStore {
"quantileMerge(0.99)(p99_duration) AS p99_duration, " +
"countIfMerge(running_count) AS active_count " +
"FROM " + view +
" WHERE tenant_id = " + lit(TENANT) +
" WHERE tenant_id = " + lit(tenantId) +
" AND bucket >= " + lit(from) +
" AND bucket < " + lit(to);
for (Filter f : filters) {
@@ -499,7 +499,7 @@ public class ClickHouseStatsStore implements StatsStore {
rs.getLong("total_count"), rs.getLong("failed_count"),
(long) rs.getDouble("avg_duration"), (long) rs.getDouble("p99_duration"),
rs.getLong("active_count")
}, TENANT, ts(from), ts(to), routeId, processorType);
}, tenantId, ts(from), ts(to), routeId, processorType);
if (!currentResult.isEmpty()) {
long[] r = currentResult.get(0);
totalCount = r[0]; failedCount = r[1]; avgDuration = r[2]; p99Duration = r[3];
@@ -511,7 +511,7 @@ public class ClickHouseStatsStore implements StatsStore {
var prevResult = jdbc.query(sql, (rs, rowNum) -> new long[]{
rs.getLong("total_count"), rs.getLong("failed_count"),
(long) rs.getDouble("avg_duration"), (long) rs.getDouble("p99_duration")
}, TENANT, ts(prevFrom), ts(prevTo), routeId, processorType);
}, tenantId, ts(prevFrom), ts(prevTo), routeId, processorType);
if (!prevResult.isEmpty()) {
long[] r = prevResult.get(0);
prevTotal = r[0]; prevFailed = r[1]; prevAvg = r[2]; prevP99 = r[3];
@@ -520,7 +520,7 @@ public class ClickHouseStatsStore implements StatsStore {
Instant todayStart = Instant.now().truncatedTo(ChronoUnit.DAYS);
long totalToday = 0;
var todayResult = jdbc.query(sql, (rs, rowNum) -> rs.getLong("total_count"),
TENANT, ts(todayStart), ts(Instant.now()), routeId, processorType);
tenantId, ts(todayStart), ts(Instant.now()), routeId, processorType);
if (!todayResult.isEmpty()) totalToday = todayResult.get(0);
return new ExecutionStats(
@@ -555,7 +555,7 @@ public class ClickHouseStatsStore implements StatsStore {
rs.getLong("total_count"), rs.getLong("failed_count"),
(long) rs.getDouble("avg_duration"), (long) rs.getDouble("p99_duration"),
rs.getLong("active_count")
), TENANT, ts(from), ts(to), routeId, processorType);
), tenantId, ts(from), ts(to), routeId, processorType);
return new StatsTimeseries(buckets);
}

19
cameleer3-server-app/src/main/java/com/cameleer3/server/app/storage/ClickHouseUsageTracker.java
View File

@@ -17,10 +17,12 @@ public class ClickHouseUsageTracker implements UsageTracker {
private static final Logger log = LoggerFactory.getLogger(ClickHouseUsageTracker.class);
private final String tenantId;
private final JdbcTemplate jdbc;
private final WriteBuffer<UsageEvent> buffer;
public ClickHouseUsageTracker(JdbcTemplate jdbc, WriteBuffer<UsageEvent> buffer) {
public ClickHouseUsageTracker(String tenantId, JdbcTemplate jdbc, WriteBuffer<UsageEvent> buffer) {
this.tenantId = tenantId;
this.jdbc = jdbc;
this.buffer = buffer;
}
@@ -35,11 +37,12 @@ public class ClickHouseUsageTracker implements UsageTracker {
if (batch.isEmpty()) return;
jdbc.batchUpdate("""
INSERT INTO usage_events (timestamp, username, method, path, normalized,
INSERT INTO usage_events (tenant_id, timestamp, username, method, path, normalized,
status_code, duration_ms, query_params)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
batch.stream().map(e -> new Object[]{
tenantId,
Timestamp.from(e.timestamp()),
e.username(),
e.method(),
@@ -59,9 +62,10 @@ public class ClickHouseUsageTracker implements UsageTracker {
count() AS cnt,
avg(duration_ms) AS avg_dur
FROM usage_events
WHERE timestamp >= ? AND timestamp < ?
WHERE tenant_id = ? AND timestamp >= ? AND timestamp < ?
""");
List<Object> params = new ArrayList<>();
params.add(tenantId);
params.add(Timestamp.from(from));
params.add(Timestamp.from(to));
@@ -80,13 +84,13 @@ public class ClickHouseUsageTracker implements UsageTracker {
String sql = """
SELECT username AS key, count() AS cnt, avg(duration_ms) AS avg_dur
FROM usage_events
WHERE timestamp >= ? AND timestamp < ?
WHERE tenant_id = ? AND timestamp >= ? AND timestamp < ?
GROUP BY key ORDER BY cnt DESC LIMIT 100
""";
return jdbc.query(sql, (rs, i) -> new UsageStats(
rs.getString("key"), rs.getLong("cnt"), rs.getLong("avg_dur")),
Timestamp.from(from), Timestamp.from(to));
tenantId, Timestamp.from(from), Timestamp.from(to));
}
public List<UsageStats> queryByHour(Instant from, Instant to, String username) {
@@ -95,9 +99,10 @@ public class ClickHouseUsageTracker implements UsageTracker {
count() AS cnt,
avg(duration_ms) AS avg_dur
FROM usage_events
WHERE timestamp >= ? AND timestamp < ?
WHERE tenant_id = ? AND timestamp >= ? AND timestamp < ?
""");
List<Object> params = new ArrayList<>();
params.add(tenantId);
params.add(Timestamp.from(from));
params.add(Timestamp.from(to));

2
cameleer3-server-app/src/main/resources/application.yml
View File

@@ -38,6 +38,8 @@ ingestion:
flush-interval-ms: 5000
cameleer:
tenant:
id: ${CAMELEER_TENANT_ID:default}
body-size-limit: ${CAMELEER_BODY_SIZE_LIMIT:16384}
indexer:
debounce-ms: ${CAMELEER_INDEXER_DEBOUNCE_MS:2000}

52
cameleer3-server-app/src/main/resources/clickhouse/init.sql
View File

@@ -6,6 +6,7 @@
CREATE TABLE IF NOT EXISTS agent_metrics (
tenant_id LowCardinality(String) DEFAULT 'default',
collected_at DateTime64(3),
environment LowCardinality(String) DEFAULT 'default',
instance_id LowCardinality(String),
metric_name LowCardinality(String),
metric_value Float64,
@@ -14,7 +15,7 @@ CREATE TABLE IF NOT EXISTS agent_metrics (
)
ENGINE = MergeTree()
PARTITION BY (tenant_id, toYYYYMM(collected_at))
ORDER BY (tenant_id, instance_id, metric_name, collected_at)
ORDER BY (tenant_id, collected_at, environment, instance_id, metric_name)
TTL toDateTime(collected_at) + INTERVAL 365 DAY DELETE
SETTINGS index_granularity = 8192;
@@ -28,6 +29,7 @@ CREATE TABLE IF NOT EXISTS executions (
route_id LowCardinality(String),
instance_id LowCardinality(String),
application_id LowCardinality(String),
environment LowCardinality(String) DEFAULT 'default',
status LowCardinality(String),
correlation_id String DEFAULT '',
exchange_id String DEFAULT '',
@@ -68,7 +70,7 @@ CREATE TABLE IF NOT EXISTS executions (
)
ENGINE = ReplacingMergeTree(_version)
PARTITION BY (tenant_id, toYYYYMM(start_time))
ORDER BY (tenant_id, start_time, application_id, route_id, execution_id)
ORDER BY (tenant_id, start_time, environment, application_id, route_id, execution_id)
TTL toDateTime(start_time) + INTERVAL 365 DAY DELETE
SETTINGS index_granularity = 8192;
@@ -85,6 +87,7 @@ CREATE TABLE IF NOT EXISTS processor_executions (
start_time DateTime64(3),
route_id LowCardinality(String),
application_id LowCardinality(String),
environment LowCardinality(String) DEFAULT 'default',
iteration Nullable(Int32),
iteration_size Nullable(Int32),
status LowCardinality(String),
@@ -116,7 +119,7 @@ CREATE TABLE IF NOT EXISTS processor_executions (
)
ENGINE = MergeTree()
PARTITION BY (tenant_id, toYYYYMM(start_time))
ORDER BY (tenant_id, start_time, application_id, route_id, execution_id, seq)
ORDER BY (tenant_id, start_time, environment, application_id, route_id, execution_id, seq)
TTL toDateTime(start_time) + INTERVAL 365 DAY DELETE
SETTINGS index_granularity = 8192;
@@ -127,6 +130,7 @@ SETTINGS index_granularity = 8192;
CREATE TABLE IF NOT EXISTS stats_1m_all (
tenant_id LowCardinality(String),
bucket DateTime,
environment LowCardinality(String) DEFAULT 'default',
total_count AggregateFunction(count),
failed_count AggregateFunction(countIf, UInt8),
running_count AggregateFunction(countIf, UInt8),
@@ -136,13 +140,14 @@ CREATE TABLE IF NOT EXISTS stats_1m_all (
)
ENGINE = AggregatingMergeTree()
PARTITION BY (tenant_id, toYYYYMM(bucket))
ORDER BY (tenant_id, bucket)
ORDER BY (tenant_id, bucket, environment)
TTL bucket + INTERVAL 365 DAY DELETE;
CREATE MATERIALIZED VIEW IF NOT EXISTS stats_1m_all_mv TO stats_1m_all AS
SELECT
tenant_id,
toStartOfMinute(start_time) AS bucket,
environment,
countState() AS total_count,
countIfState(status = 'FAILED') AS failed_count,
countIfState(status = 'RUNNING') AS running_count,
@@ -150,7 +155,7 @@ SELECT
maxState(duration_ms) AS duration_max,
quantileState(0.99)(duration_ms) AS p99_duration
FROM executions
GROUP BY tenant_id, bucket;
GROUP BY tenant_id, bucket, environment;
-- stats_1m_app (per-application)
@@ -158,6 +163,7 @@ CREATE TABLE IF NOT EXISTS stats_1m_app (
tenant_id LowCardinality(String),
application_id LowCardinality(String),
bucket DateTime,
environment LowCardinality(String) DEFAULT 'default',
total_count AggregateFunction(count),
failed_count AggregateFunction(countIf, UInt8),
running_count AggregateFunction(countIf, UInt8),
@@ -167,7 +173,7 @@ CREATE TABLE IF NOT EXISTS stats_1m_app (
)
ENGINE = AggregatingMergeTree()
PARTITION BY (tenant_id, toYYYYMM(bucket))
ORDER BY (tenant_id, application_id, bucket)
ORDER BY (tenant_id, bucket, environment, application_id)
TTL bucket + INTERVAL 365 DAY DELETE;
CREATE MATERIALIZED VIEW IF NOT EXISTS stats_1m_app_mv TO stats_1m_app AS
@@ -175,6 +181,7 @@ SELECT
tenant_id,
application_id,
toStartOfMinute(start_time) AS bucket,
environment,
countState() AS total_count,
countIfState(status = 'FAILED') AS failed_count,
countIfState(status = 'RUNNING') AS running_count,
@@ -182,7 +189,7 @@ SELECT
maxState(duration_ms) AS duration_max,
quantileState(0.99)(duration_ms) AS p99_duration
FROM executions
GROUP BY tenant_id, application_id, bucket;
GROUP BY tenant_id, application_id, bucket, environment;
-- stats_1m_route (per-route)
@@ -191,6 +198,7 @@ CREATE TABLE IF NOT EXISTS stats_1m_route (
application_id LowCardinality(String),
route_id LowCardinality(String),
bucket DateTime,
environment LowCardinality(String) DEFAULT 'default',
total_count AggregateFunction(count),
failed_count AggregateFunction(countIf, UInt8),
running_count AggregateFunction(countIf, UInt8),
@@ -200,7 +208,7 @@ CREATE TABLE IF NOT EXISTS stats_1m_route (
)
ENGINE = AggregatingMergeTree()
PARTITION BY (tenant_id, toYYYYMM(bucket))
ORDER BY (tenant_id, application_id, route_id, bucket)
ORDER BY (tenant_id, bucket, environment, application_id, route_id)
TTL bucket + INTERVAL 365 DAY DELETE;
CREATE MATERIALIZED VIEW IF NOT EXISTS stats_1m_route_mv TO stats_1m_route AS
@@ -209,6 +217,7 @@ SELECT
application_id,
route_id,
toStartOfMinute(start_time) AS bucket,
environment,
countState() AS total_count,
countIfState(status = 'FAILED') AS failed_count,
countIfState(status = 'RUNNING') AS running_count,
@@ -216,7 +225,7 @@ SELECT
maxState(duration_ms) AS duration_max,
quantileState(0.99)(duration_ms) AS p99_duration
FROM executions
GROUP BY tenant_id, application_id, route_id, bucket;
GROUP BY tenant_id, application_id, route_id, bucket, environment;
-- stats_1m_processor (per-processor-type)
@@ -225,6 +234,7 @@ CREATE TABLE IF NOT EXISTS stats_1m_processor (
application_id LowCardinality(String),
processor_type LowCardinality(String),
bucket DateTime,
environment LowCardinality(String) DEFAULT 'default',
total_count AggregateFunction(count),
failed_count AggregateFunction(countIf, UInt8),
duration_sum AggregateFunction(sum, Nullable(Int64)),
@@ -233,7 +243,7 @@ CREATE TABLE IF NOT EXISTS stats_1m_processor (
)
ENGINE = AggregatingMergeTree()
PARTITION BY (tenant_id, toYYYYMM(bucket))
ORDER BY (tenant_id, application_id, processor_type, bucket)
ORDER BY (tenant_id, bucket, environment, application_id, processor_type)
TTL bucket + INTERVAL 365 DAY DELETE;
CREATE MATERIALIZED VIEW IF NOT EXISTS stats_1m_processor_mv TO stats_1m_processor AS
@@ -242,13 +252,14 @@ SELECT
application_id,
processor_type,
toStartOfMinute(start_time) AS bucket,
environment,
countState() AS total_count,
countIfState(status = 'FAILED') AS failed_count,
sumState(duration_ms) AS duration_sum,
maxState(duration_ms) AS duration_max,
quantileState(0.99)(duration_ms) AS p99_duration
FROM processor_executions
GROUP BY tenant_id, application_id, processor_type, bucket;
GROUP BY tenant_id, application_id, processor_type, bucket, environment;
-- stats_1m_processor_detail (per-processor-id)
@@ -259,6 +270,7 @@ CREATE TABLE IF NOT EXISTS stats_1m_processor_detail (
processor_id String,
processor_type LowCardinality(String),
bucket DateTime,
environment LowCardinality(String) DEFAULT 'default',
total_count AggregateFunction(count),
failed_count AggregateFunction(countIf, UInt8),
duration_sum AggregateFunction(sum, Nullable(Int64)),
@@ -267,7 +279,7 @@ CREATE TABLE IF NOT EXISTS stats_1m_processor_detail (
)
ENGINE = AggregatingMergeTree()
PARTITION BY (tenant_id, toYYYYMM(bucket))
ORDER BY (tenant_id, application_id, route_id, processor_id, processor_type, bucket)
ORDER BY (tenant_id, bucket, environment, application_id, route_id, processor_id, processor_type)
TTL bucket + INTERVAL 365 DAY DELETE;
CREATE MATERIALIZED VIEW IF NOT EXISTS stats_1m_processor_detail_mv TO stats_1m_processor_detail AS
@@ -278,13 +290,14 @@ SELECT
processor_id,
processor_type,
toStartOfMinute(start_time) AS bucket,
environment,
countState() AS total_count,
countIfState(status = 'FAILED') AS failed_count,
sumState(duration_ms) AS duration_sum,
maxState(duration_ms) AS duration_max,
quantileState(0.99)(duration_ms) AS p99_duration
FROM processor_executions
GROUP BY tenant_id, application_id, route_id, processor_id, processor_type, bucket;
GROUP BY tenant_id, application_id, route_id, processor_id, processor_type, bucket, environment;
-- ── Route Diagrams ──────────────────────────────────────────────────────
@@ -294,11 +307,12 @@ CREATE TABLE IF NOT EXISTS route_diagrams (
route_id LowCardinality(String),
instance_id LowCardinality(String),
application_id LowCardinality(String),
environment LowCardinality(String) DEFAULT 'default',
definition String,
created_at DateTime64(3) DEFAULT now64(3)
)
ENGINE = ReplacingMergeTree(created_at)
ORDER BY (tenant_id, content_hash)
ORDER BY (tenant_id, environment, route_id, instance_id, content_hash)
SETTINGS index_granularity = 8192;
-- ── Agent Events ────────────────────────────────────────────────────────
@@ -306,6 +320,7 @@ SETTINGS index_granularity = 8192;
CREATE TABLE IF NOT EXISTS agent_events (
tenant_id LowCardinality(String) DEFAULT 'default',
timestamp DateTime64(3) DEFAULT now64(3),
environment LowCardinality(String) DEFAULT 'default',
instance_id LowCardinality(String),
application_id LowCardinality(String),
event_type LowCardinality(String),
@@ -313,7 +328,7 @@ CREATE TABLE IF NOT EXISTS agent_events (
)
ENGINE = MergeTree()
PARTITION BY (tenant_id, toYYYYMM(timestamp))
ORDER BY (tenant_id, application_id, instance_id, timestamp)
ORDER BY (tenant_id, timestamp, environment, instance_id)
TTL toDateTime(timestamp) + INTERVAL 365 DAY DELETE;
-- ── Logs ────────────────────────────────────────────────────────────────
@@ -321,6 +336,7 @@ TTL toDateTime(timestamp) + INTERVAL 365 DAY DELETE;
CREATE TABLE IF NOT EXISTS logs (
tenant_id LowCardinality(String) DEFAULT 'default',
timestamp DateTime64(3),
environment LowCardinality(String) DEFAULT 'default',
application LowCardinality(String),
instance_id LowCardinality(String),
level LowCardinality(String),
@@ -337,14 +353,16 @@ CREATE TABLE IF NOT EXISTS logs (
)
ENGINE = MergeTree()
PARTITION BY (tenant_id, toYYYYMM(timestamp))
ORDER BY (tenant_id, application, timestamp)
ORDER BY (tenant_id, timestamp, environment, application, instance_id)
TTL toDateTime(timestamp) + INTERVAL 365 DAY DELETE
SETTINGS index_granularity = 8192;
-- ── Usage Events ────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS usage_events (
tenant_id LowCardinality(String) DEFAULT 'default',
timestamp DateTime64(3) DEFAULT now64(3),
environment LowCardinality(String) DEFAULT 'default',
username LowCardinality(String),
method LowCardinality(String),
path String,
@@ -354,5 +372,5 @@ CREATE TABLE IF NOT EXISTS usage_events (
query_params String DEFAULT ''
)
ENGINE = MergeTree()
ORDER BY (username, timestamp)
ORDER BY (tenant_id, timestamp, environment, username, normalized)
TTL toDateTime(timestamp) + INTERVAL 90 DAY;