|
@@ -0,0 +1,228 @@
|
|
|
|
|
+package cn.seecoder.analysis.d4;
|
|
|
|
|
+
|
|
|
|
|
+import cn.seecoder.analysis.config.D4MetricsProperties;
|
|
|
|
|
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
|
|
|
|
|
+import co.elastic.clients.elasticsearch._types.FieldValue;
|
|
|
|
|
+import co.elastic.clients.elasticsearch._types.SortOrder;
|
|
|
|
|
+import co.elastic.clients.elasticsearch._types.query_dsl.Query;
|
|
|
|
|
+import co.elastic.clients.elasticsearch.core.SearchResponse;
|
|
|
|
|
+import co.elastic.clients.elasticsearch.core.search.Hit;
|
|
|
|
|
+import co.elastic.clients.json.JsonData;
|
|
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
|
|
+
|
|
|
|
|
+import java.io.IOException;
|
|
|
|
|
+import java.util.LinkedHashMap;
|
|
|
|
|
+import java.util.List;
|
|
|
|
|
+import java.util.Locale;
|
|
|
|
|
+import java.util.Map;
|
|
|
|
|
+import java.util.Objects;
|
|
|
|
|
+
|
|
|
|
|
+@Service
|
|
|
|
|
+public class D4MetricsService {
|
|
|
|
|
+
|
|
|
|
|
+ private static final List<String> USER_FIELDS = List.of(
|
|
|
|
|
+ "creatorId",
|
|
|
|
|
+ "receiverId",
|
|
|
|
|
+ "reviewerId",
|
|
|
|
|
+ "operatorId",
|
|
|
|
|
+ "resolverId"
|
|
|
|
|
+ );
|
|
|
|
|
+
|
|
|
|
|
+ private final ElasticsearchClient elasticsearchClient;
|
|
|
|
|
+ private final D4MetricsProperties properties;
|
|
|
|
|
+
|
|
|
|
|
+ public D4MetricsService(ElasticsearchClient elasticsearchClient, D4MetricsProperties properties) {
|
|
|
|
|
+ this.elasticsearchClient = elasticsearchClient;
|
|
|
|
|
+ this.properties = properties;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ public D4MetricsResponse calculate(D4MetricQuery metricQuery) {
|
|
|
|
|
+ AggregationState state = new AggregationState(metricQuery);
|
|
|
|
|
+ int scanned = 0;
|
|
|
|
|
+ int offset = 0;
|
|
|
|
|
+ boolean truncated = false;
|
|
|
|
|
+
|
|
|
|
|
+ while (scanned < properties.getMaxEventsToScan()) {
|
|
|
|
|
+ int size = Math.min(properties.getPageSize(), properties.getMaxEventsToScan() - scanned);
|
|
|
|
|
+ SearchResponse<LogEvent> response = search(metricQuery, offset, size);
|
|
|
|
|
+ List<Hit<LogEvent>> hits = response.hits().hits();
|
|
|
|
|
+ if (hits.isEmpty()) {
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ for (Hit<LogEvent> hit : hits) {
|
|
|
|
|
+ LogEvent event = hit.source();
|
|
|
|
|
+ if (event == null) {
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+ scanned++;
|
|
|
|
|
+ state.accept(event);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (hits.size() < size) {
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+ offset += hits.size();
|
|
|
|
|
+ truncated = scanned >= properties.getMaxEventsToScan();
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return state.toResponse(properties.getIndexPattern(), scanned, properties.getMaxEventsToScan(), truncated);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private SearchResponse<LogEvent> search(D4MetricQuery metricQuery, int from, int size) {
|
|
|
|
|
+ Query query = Query.of(q -> q.bool(b -> b
|
|
|
|
|
+ .filter(f -> f.range(r -> r
|
|
|
|
|
+ .field("@timestamp")
|
|
|
|
|
+ .gte(JsonData.of(metricQuery.from().toString()))
|
|
|
|
|
+ .lt(JsonData.of(metricQuery.to().toString()))))
|
|
|
|
|
+ .filter(f -> f.terms(t -> t
|
|
|
|
|
+ .field("bizName.keyword")
|
|
|
|
|
+ .terms(v -> v.value(D4EventType.ALL_BIZ_NAMES.stream()
|
|
|
|
|
+ .map(FieldValue::of)
|
|
|
|
|
+ .toList()))))));
|
|
|
|
|
+
|
|
|
|
|
+ try {
|
|
|
|
|
+ return elasticsearchClient.search(s -> s
|
|
|
|
|
+ .index(properties.getIndexPattern())
|
|
|
|
|
+ .ignoreUnavailable(true)
|
|
|
|
|
+ .allowNoIndices(true)
|
|
|
|
|
+ .from(from)
|
|
|
|
|
+ .size(size)
|
|
|
|
|
+ .query(query)
|
|
|
|
|
+ .sort(sort -> sort.field(f -> f.field("@timestamp").order(SortOrder.Asc))),
|
|
|
|
|
+ LogEvent.class);
|
|
|
|
|
+ } catch (IOException ex) {
|
|
|
|
|
+ throw new D4MetricsQueryException("Failed to query D4 logs from Elasticsearch", ex);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private static final class AggregationState {
|
|
|
|
|
+
|
|
|
|
|
+ private final D4MetricQuery query;
|
|
|
|
|
+ private final Map<String, Long> eventCounts = new LinkedHashMap<>();
|
|
|
|
|
+ private long assignedCount;
|
|
|
|
|
+ private long completedCount;
|
|
|
|
|
+ private long onTimeCompletedCount;
|
|
|
|
|
+ private long sentMessageCount;
|
|
|
|
|
+ private long respondedMessageCount;
|
|
|
|
|
+ private long responseMsSum;
|
|
|
|
|
+ private long effectiveCollaborationCount;
|
|
|
|
|
+ private long conflictCreatedCount;
|
|
|
|
|
+ private long conflictResolvedCount;
|
|
|
|
|
+
|
|
|
|
|
+ private AggregationState(D4MetricQuery query) {
|
|
|
|
|
+ this.query = query;
|
|
|
|
|
+ D4EventType.ALL_BIZ_NAMES.stream()
|
|
|
|
|
+ .sorted()
|
|
|
|
|
+ .forEach(bizName -> eventCounts.put(bizName, 0L));
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private void accept(LogEvent event) {
|
|
|
|
|
+ if (isDuplicateConflictTroubleshootingLog(event)) {
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ Map<String, String> businessInfo = event.parsedBusinessInfo();
|
|
|
|
|
+ if (!matchesFilters(businessInfo)) {
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ eventCounts.computeIfPresent(event.getBizName(), (key, value) -> value + 1);
|
|
|
|
|
+
|
|
|
|
|
+ if (D4EventType.TASK_ASSIGNED.bizName().equals(event.getBizName())) {
|
|
|
|
|
+ assignedCount++;
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (D4EventType.TASK_COMPLETED.bizName().equals(event.getBizName())) {
|
|
|
|
|
+ completedCount++;
|
|
|
|
|
+ Long taskDurationMs = BusinessInfoParser.longValue(businessInfo, "taskDurationMs");
|
|
|
|
|
+ if (query.slaMs() != null && taskDurationMs != null && taskDurationMs <= query.slaMs()) {
|
|
|
|
|
+ onTimeCompletedCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (D4EventType.COMM_MESSAGE_SENT.bizName().equals(event.getBizName())) {
|
|
|
|
|
+ sentMessageCount++;
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (D4EventType.COMM_MESSAGE_RESPONSE.bizName().equals(event.getBizName())) {
|
|
|
|
|
+ Long responseMs = BusinessInfoParser.longValue(businessInfo, "responseMs");
|
|
|
|
|
+ if (responseMs != null) {
|
|
|
|
|
+ respondedMessageCount++;
|
|
|
|
|
+ responseMsSum += responseMs;
|
|
|
|
|
+ }
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (D4EventType.COLLAB_EFFECTIVE.bizName().equals(event.getBizName())) {
|
|
|
|
|
+ effectiveCollaborationCount++;
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (D4EventType.COLLAB_CONFLICT_CREATED.bizName().equals(event.getBizName())) {
|
|
|
|
|
+ conflictCreatedCount++;
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (D4EventType.COLLAB_CONFLICT_RESOLVED.bizName().equals(event.getBizName())) {
|
|
|
|
|
+ conflictResolvedCount++;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private boolean matchesFilters(Map<String, String> businessInfo) {
|
|
|
|
|
+ if (query.projectId() != null && !Objects.equals(query.projectId(), BusinessInfoParser.longValue(businessInfo, "projectId"))) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (query.taskType() != null) {
|
|
|
|
|
+ String taskType = businessInfo.get("taskType");
|
|
|
|
|
+ if (taskType == null || !taskType.equalsIgnoreCase(query.taskType())) {
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if (query.userId() != null) {
|
|
|
|
|
+ return USER_FIELDS.stream()
|
|
|
|
|
+ .map(field -> BusinessInfoParser.longValue(businessInfo, field))
|
|
|
|
|
+ .anyMatch(query.userId()::equals);
|
|
|
|
|
+ }
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private boolean isDuplicateConflictTroubleshootingLog(LogEvent event) {
|
|
|
|
|
+ return D4EventType.COLLAB_CONFLICT_CREATED.bizName().equals(event.getBizName())
|
|
|
|
|
+ && event.getLogLevel() != null
|
|
|
|
|
+ && "ERROR".equals(event.getLogLevel().toUpperCase(Locale.ROOT));
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private D4MetricsResponse toResponse(String indexPattern, int scanned, int maxEventsToScan, boolean truncated) {
|
|
|
|
|
+ return new D4MetricsResponse(
|
|
|
|
|
+ query.from(),
|
|
|
|
|
+ query.to(),
|
|
|
|
|
+ new D4MetricsResponse.Filter(query.projectId(), query.taskType(), query.userId(), query.slaMs()),
|
|
|
|
|
+ new D4MetricsResponse.TaskCompletionMetric(
|
|
|
|
|
+ assignedCount,
|
|
|
|
|
+ completedCount,
|
|
|
|
|
+ rate(completedCount, assignedCount),
|
|
|
|
|
+ query.slaMs() == null ? null : onTimeCompletedCount,
|
|
|
|
|
+ query.slaMs() == null ? null : rate(onTimeCompletedCount, completedCount)
|
|
|
|
|
+ ),
|
|
|
|
|
+ new D4MetricsResponse.CommunicationMetric(
|
|
|
|
|
+ sentMessageCount,
|
|
|
|
|
+ respondedMessageCount,
|
|
|
|
|
+ respondedMessageCount == 0 ? null : (double) responseMsSum / respondedMessageCount
|
|
|
|
|
+ ),
|
|
|
|
|
+ new D4MetricsResponse.CollaborationMetric(effectiveCollaborationCount),
|
|
|
|
|
+ new D4MetricsResponse.ConflictMetric(
|
|
|
|
|
+ conflictCreatedCount,
|
|
|
|
|
+ conflictResolvedCount,
|
|
|
|
|
+ rate(conflictResolvedCount, conflictCreatedCount)
|
|
|
|
|
+ ),
|
|
|
|
|
+ eventCounts,
|
|
|
|
|
+ new D4MetricsResponse.ScanInfo(indexPattern, scanned, maxEventsToScan, truncated)
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ private Double rate(long numerator, long denominator) {
|
|
|
|
|
+ if (denominator == 0) {
|
|
|
|
|
+ return null;
|
|
|
|
|
+ }
|
|
|
|
|
+ return (double) numerator / denominator;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|