From 0b82c956d826d20e538f546d3b449d26adeaeb05 Mon Sep 17 00:00:00 2001 From: Aliaska Varieva Date: Mon, 14 Sep 2026 20:50:56 +0200 Subject: [PATCH 1/3] feat(devices): device agent logs API backed by Loki Add the openframe-data-loki module (Loki query client over Spring's RestClient HTTP interface, LogQL escaping) and a deviceLogs GraphQL query that reads the agent-logs streams written by openframe-saas-logs-stream. The stream selector is pinned to the tenant's own domain from the tenants collection and the device must be visible to the tenant. Results are newest first with a timestamp+skip cursor. Disabled unless openframe.loki.enabled=true. --- openframe-api-lib/pom.xml | 4 + .../api/dto/device/DeviceLogEntry.java | 41 ++++ .../dto/device/DeviceLogFilterCriteria.java | 36 +++ .../api/dto/device/DeviceLogLevel.java | 11 + .../api/service/device/DeviceLogCursor.java | 37 +++ .../api/service/device/DeviceLogService.java | 223 ++++++++++++++++++ .../service/device/DeviceLogServiceTest.java | 197 ++++++++++++++++ .../api/datafetcher/DeviceLogDataFetcher.java | 44 ++++ .../api/dto/device/DeviceLogFilterInput.java | 26 ++ .../exception/GraphQLExceptionHandler.java | 3 + .../api/mapper/GraphQLDeviceLogMapper.java | 56 +++++ .../main/resources/schema/device-log.graphqls | 43 ++++ .../mapper/GraphQLDeviceLogMapperTest.java | 28 +++ openframe-data-loki/pom.xml | 39 +++ .../com/openframe/data/loki/client/LogQl.java | 52 ++++ .../data/loki/client/LokiClient.java | 90 +++++++ .../data/loki/client/LokiHttpApi.java | 24 ++ .../data/loki/client/LokiQueryException.java | 12 + .../data/loki/config/LokiConfig.java | 37 +++ .../data/loki/config/LokiProperties.java | 31 +++ .../data/loki/model/LokiDirection.java | 6 + .../data/loki/model/LokiLogEntry.java | 14 ++ .../data/loki/model/LokiQueryResponse.java | 23 ++ .../openframe/data/loki/client/LogQlTest.java | 24 ++ .../data/loki/client/LokiClientTest.java | 97 ++++++++ .../openframe/core/exception/ErrorCode.java | 1 + pom.xml | 6 + 27 files changed, 1205 insertions(+) create mode 100644 openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogEntry.java create mode 100644 openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogFilterCriteria.java create mode 100644 openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogLevel.java create mode 100644 openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogCursor.java create mode 100644 openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java create mode 100644 openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java create mode 100644 openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/DeviceLogDataFetcher.java create mode 100644 openframe-api-service-core/src/main/java/com/openframe/api/dto/device/DeviceLogFilterInput.java create mode 100644 openframe-api-service-core/src/main/java/com/openframe/api/mapper/GraphQLDeviceLogMapper.java create mode 100644 openframe-api-service-core/src/main/resources/schema/device-log.graphqls create mode 100644 openframe-api-service-core/src/test/java/com/openframe/api/mapper/GraphQLDeviceLogMapperTest.java create mode 100644 openframe-data-loki/pom.xml create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/client/LogQl.java create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiClient.java create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiHttpApi.java create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiQueryException.java create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/config/LokiConfig.java create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/config/LokiProperties.java create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiDirection.java create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiLogEntry.java create mode 100644 openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiQueryResponse.java create mode 100644 openframe-data-loki/src/test/java/com/openframe/data/loki/client/LogQlTest.java create mode 100644 openframe-data-loki/src/test/java/com/openframe/data/loki/client/LokiClientTest.java diff --git a/openframe-api-lib/pom.xml b/openframe-api-lib/pom.xml index 7818777252..e435d4e41a 100644 --- a/openframe-api-lib/pom.xml +++ b/openframe-api-lib/pom.xml @@ -35,6 +35,10 @@ com.openframe.oss openframe-data-pinot + + com.openframe.oss + openframe-data-loki + diff --git a/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogEntry.java b/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogEntry.java new file mode 100644 index 0000000000..7dbcd161d5 --- /dev/null +++ b/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogEntry.java @@ -0,0 +1,41 @@ +package com.openframe.api.dto.device; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.time.Instant; + +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class DeviceLogEntry { + + /** + * When the logs pipeline ingested the entry, nanosecond precision. The sort key. + */ + private Instant timestamp; + + /** + * When the agent wrote the line, as reported by the agent. + */ + private Instant agentTimestamp; + + private String level; + + private String message; + + private String hostname; + + /** + * Number of identical lines the agent collapsed into this entry. + */ + private Long count; + + /** + * Opaque cursor positioned at this entry. + */ + private String cursor; +} diff --git a/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogFilterCriteria.java b/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogFilterCriteria.java new file mode 100644 index 0000000000..7b6a4ce758 --- /dev/null +++ b/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogFilterCriteria.java @@ -0,0 +1,36 @@ +package com.openframe.api.dto.device; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.time.Instant; +import java.util.List; + +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class DeviceLogFilterCriteria { + + /** + * Any of these levels; all levels when empty. + */ + private List levels; + + /** + * Case-insensitive substring match on the log message. + */ + private String search; + + /** + * Inclusive lower bound; defaults to {@code to} minus the default lookback. + */ + private Instant from; + + /** + * Inclusive upper bound; defaults to now. + */ + private Instant to; +} diff --git a/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogLevel.java b/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogLevel.java new file mode 100644 index 0000000000..41fb76ea16 --- /dev/null +++ b/openframe-api-lib/src/main/java/com/openframe/api/dto/device/DeviceLogLevel.java @@ -0,0 +1,11 @@ +package com.openframe.api.dto.device; + +/** + * Agent log levels, as carried by the {@code level} label of the agent-logs Loki stream. + */ +public enum DeviceLogLevel { + DEBUG, + INFO, + WARN, + ERROR +} diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogCursor.java b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogCursor.java new file mode 100644 index 0000000000..f69277bf17 --- /dev/null +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogCursor.java @@ -0,0 +1,37 @@ +package com.openframe.api.service.device; + +import com.openframe.api.dto.shared.CursorCodec; + +/** + * Position in a newest-first device log listing: the Loki timestamp of the last returned entry and how many + * entries at exactly that timestamp have been returned so far. Loki has no offsets, so the next page is read + * up to and including that timestamp and the already returned entries at it are skipped. + */ +record DeviceLogCursor(long timestampNanos, int skip) { + + static final int MAX_SKIP = 1000; + + String encode() { + return CursorCodec.encode(timestampNanos + ":" + skip); + } + + /** + * Parses a cursor already base64-decoded by {@code CursorPaginationCriteria}; {@code null} when absent. + */ + static DeviceLogCursor fromRaw(String raw) { + if (raw == null) { + return null; + } + try { + int separator = raw.indexOf(':'); + long timestampNanos = Long.parseLong(raw.substring(0, separator)); + int skip = Integer.parseInt(raw.substring(separator + 1)); + if (timestampNanos > 0 && skip >= 0 && skip <= MAX_SKIP) { + return new DeviceLogCursor(timestampNanos, skip); + } + } catch (RuntimeException ignored) { + // fall through to the rejection below + } + throw new IllegalArgumentException("Invalid cursor"); + } +} diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java new file mode 100644 index 0000000000..c42dfd21af --- /dev/null +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java @@ -0,0 +1,223 @@ +package com.openframe.api.service.device; + +import com.openframe.api.dto.GenericQueryResult; +import com.openframe.api.dto.device.DeviceLogEntry; +import com.openframe.api.dto.device.DeviceLogFilterCriteria; +import com.openframe.api.dto.device.DeviceLogLevel; +import com.openframe.api.dto.shared.CursorPaginationCriteria; +import com.openframe.api.dto.shared.PageInfo; +import com.openframe.api.exception.DeviceNotFoundException; +import com.openframe.core.exception.InternalException; +import com.openframe.data.document.tenant.Tenant; +import com.openframe.data.loki.client.LogQl; +import com.openframe.data.loki.client.LokiClient; +import com.openframe.data.loki.model.LokiDirection; +import com.openframe.data.loki.model.LokiLogEntry; +import com.openframe.data.repository.tenant.TenantRepository; +import com.openframe.data.service.TenantIdProvider; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; + +import java.time.Duration; +import java.time.Instant; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.List; + +import static java.util.stream.Collectors.joining; + +/** + * Device agent logs, read from Loki where {@code openframe-saas-logs-stream} writes them as + * {@code {job="agent-logs", tenant_domain, level}} streams with {@code machine_id}, {@code hostname}, + * {@code agent_ts} and {@code count} as structured metadata. + */ +@Service +@Slf4j +@RequiredArgsConstructor +@ConditionalOnProperty(name = "openframe.loki.enabled", havingValue = "true") +public class DeviceLogService { + + static final Duration DEFAULT_LOOKBACK = Duration.ofDays(7); + static final Duration MAX_RANGE = Duration.ofDays(30); + static final int MAX_SEARCH_LENGTH = 256; + + private static final String AGENT_LOGS_JOB = "agent-logs"; + private static final long NANOS_PER_SECOND = 1_000_000_000L; + + private final LokiClient lokiClient; + private final DeviceService deviceService; + private final TenantIdProvider tenantIdProvider; + private final TenantRepository tenantRepository; + + /** + * Logs of one device, newest first. + *

+ * Tenant isolation does not rely on anything in the request: the device must be visible to this tenant, + * and the stream selector is pinned to this tenant's own domain from the {@code tenants} collection. + */ + public GenericQueryResult queryDeviceLogs(String machineId, + DeviceLogFilterCriteria filter, + CursorPaginationCriteria pagination) { + deviceService.findByMachineId(machineId) + .orElseThrow(() -> new DeviceNotFoundException("Machine not found: " + machineId)); + + DeviceLogFilterCriteria criteria = filter != null ? filter : new DeviceLogFilterCriteria(); + CursorPaginationCriteria page = (pagination != null ? pagination : new CursorPaginationCriteria()).normalize(); + DeviceLogCursor after = DeviceLogCursor.fromRaw(page.getCursor()); + validateSearch(criteria.getSearch()); + + Instant to = criteria.getTo() != null ? criteria.getTo() : Instant.now(); + Instant from = criteria.getFrom() != null ? criteria.getFrom() : to.minus(DEFAULT_LOOKBACK); + validateRange(from, to); + + long startNanos = toNanos(from); + // Loki's end is exclusive while both the filter's upper bound and the cursor entry are inclusive + long endNanos = toNanos(to) + 1; + int skip = 0; + if (after != null) { + if (after.timestampNanos() < startNanos) { + return result(List.of(), false, true); + } + endNanos = Math.min(endNanos, after.timestampNanos() + 1); + skip = after.skip(); + } + + String query = buildQuery(resolveTenantDomain(), machineId, criteria); + int pageSize = page.getLimit(); + log.debug("Querying device logs for machineId: {}, query: {}, start: {}, end: {}", machineId, query, startNanos, endNanos); + + // One extra entry tells whether there is a next page + List entries = lokiClient.queryRange(query, startNanos, endNanos, pageSize + skip + 1, + LokiDirection.BACKWARD); + List remaining = dropReturned(entries, after); + + return result(toItems(remaining.subList(0, Math.min(pageSize, remaining.size())), after), + remaining.size() > pageSize, after != null); + } + + static String buildQuery(String tenantDomain, String machineId, DeviceLogFilterCriteria criteria) { + StringBuilder query = new StringBuilder("{job=").append(LogQl.quote(AGENT_LOGS_JOB)) + .append(", tenant_domain=").append(LogQl.quote(tenantDomain)); + List levels = criteria.getLevels(); + if (levels != null && !levels.isEmpty()) { + String alternatives = levels.stream().distinct().map(Enum::name).collect(joining("|")); + query.append(", level=~").append(LogQl.quote(alternatives)); + } + query.append('}'); + if (StringUtils.hasText(criteria.getSearch())) { + // Line filter before the metadata filter: the cheapest stage runs first + query.append(" |~ ").append(LogQl.quote("(?i)" + LogQl.regexLiteral(criteria.getSearch()))); + } + return query.append(" | machine_id=").append(LogQl.quote(machineId)).toString(); + } + + private String resolveTenantDomain() { + String tenantId = tenantIdProvider.getTenantId(); + return tenantRepository.findById(tenantId) + .map(Tenant::getDomain) + .filter(StringUtils::hasText) + .orElseThrow(() -> { + log.error("Cannot query device logs: tenant {} has no domain", tenantId); + return new InternalException("Device logs are not available for this tenant"); + }); + } + + /** + * Drops the entries at the cursor's timestamp that earlier pages already returned; they come first + * because the query ends right after that timestamp. + */ + private static List dropReturned(List entries, DeviceLogCursor after) { + int index = 0; + if (after != null) { + while (index < entries.size() && index < after.skip() + && entries.get(index).timestampNanos() == after.timestampNanos()) { + index++; + } + } + return entries.subList(index, entries.size()); + } + + private static List toItems(List entries, DeviceLogCursor after) { + List items = new ArrayList<>(entries.size()); + long runTimestamp = after != null ? after.timestampNanos() : Long.MIN_VALUE; + int runLength = after != null ? after.skip() : 0; + for (LokiLogEntry entry : entries) { + if (entry.timestampNanos() != runTimestamp) { + runTimestamp = entry.timestampNanos(); + runLength = 0; + } + runLength++; + items.add(DeviceLogEntry.builder() + .timestamp(entry.timestamp()) + .agentTimestamp(parseInstant(entry.labels().get("agent_ts"))) + .level(entry.labels().get("level")) + .message(entry.line()) + .hostname(entry.labels().get("hostname")) + .count(parseLong(entry.labels().get("count"))) + .cursor(new DeviceLogCursor(runTimestamp, runLength).encode()) + .build()); + } + return items; + } + + private static GenericQueryResult result(List items, boolean hasNextPage, + boolean hasPreviousPage) { + return GenericQueryResult.builder() + .items(items) + .pageInfo(PageInfo.builder() + .hasNextPage(hasNextPage) + .hasPreviousPage(hasPreviousPage) + .startCursor(items.isEmpty() ? null : items.get(0).getCursor()) + .endCursor(items.isEmpty() ? null : items.get(items.size() - 1).getCursor()) + .build()) + .build(); + } + + private static void validateSearch(String search) { + if (search != null && search.length() > MAX_SEARCH_LENGTH) { + throw new IllegalArgumentException("search cannot exceed " + MAX_SEARCH_LENGTH + " characters"); + } + } + + private static void validateRange(Instant from, Instant to) { + if (!from.isBefore(to)) { + throw new IllegalArgumentException("'from' must be before 'to'"); + } + if (Duration.between(from, to).compareTo(MAX_RANGE) > 0) { + throw new IllegalArgumentException("Time range cannot exceed " + MAX_RANGE.toDays() + " days"); + } + } + + private static long toNanos(Instant instant) { + try { + return Math.addExact(Math.multiplyExact(instant.getEpochSecond(), NANOS_PER_SECOND), instant.getNano()); + } catch (ArithmeticException e) { + throw new IllegalArgumentException("Timestamp out of range: " + instant); + } + } + + private static Instant parseInstant(String value) { + if (value == null) { + return null; + } + try { + return Instant.parse(value); + } catch (DateTimeParseException e) { + return null; + } + } + + private static Long parseLong(String value) { + if (value == null) { + return null; + } + try { + return Long.valueOf(value); + } catch (NumberFormatException e) { + return null; + } + } +} diff --git a/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java b/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java new file mode 100644 index 0000000000..d88e817567 --- /dev/null +++ b/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java @@ -0,0 +1,197 @@ +package com.openframe.api.service.device; + +import com.openframe.api.dto.GenericQueryResult; +import com.openframe.api.dto.device.DeviceLogEntry; +import com.openframe.api.dto.device.DeviceLogFilterCriteria; +import com.openframe.api.dto.device.DeviceLogLevel; +import com.openframe.api.dto.shared.CursorCodec; +import com.openframe.api.dto.shared.CursorPaginationCriteria; +import com.openframe.api.exception.DeviceNotFoundException; +import com.openframe.core.exception.InternalException; +import com.openframe.data.document.device.Machine; +import com.openframe.data.document.tenant.Tenant; +import com.openframe.data.loki.client.LokiClient; +import com.openframe.data.loki.model.LokiDirection; +import com.openframe.data.loki.model.LokiLogEntry; +import com.openframe.data.repository.tenant.TenantRepository; +import com.openframe.data.service.TenantIdProvider; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +@MockitoSettings(strictness = Strictness.LENIENT) +class DeviceLogServiceTest { + + private static final String MACHINE_ID = "machine-1"; + private static final String TENANT_ID = "tenant-1"; + private static final String TENANT_DOMAIN = "acme.openframe.ai"; + private static final Instant TO = Instant.parse("2026-09-14T12:00:00Z"); + private static final Instant FROM = TO.minus(Duration.ofDays(1)); + private static final long TO_NANOS = TO.getEpochSecond() * 1_000_000_000L; + private static final long FROM_NANOS = FROM.getEpochSecond() * 1_000_000_000L; + + @Mock private LokiClient lokiClient; + @Mock private DeviceService deviceService; + @Mock private TenantIdProvider tenantIdProvider; + @Mock private TenantRepository tenantRepository; + + private DeviceLogService service; + + @BeforeEach + void setUp() { + service = new DeviceLogService(lokiClient, deviceService, tenantIdProvider, tenantRepository); + when(deviceService.findByMachineId(MACHINE_ID)).thenReturn(Optional.of(mock(Machine.class))); + when(tenantIdProvider.getTenantId()).thenReturn(TENANT_ID); + when(tenantRepository.findById(TENANT_ID)) + .thenReturn(Optional.of(Tenant.builder().id(TENANT_ID).domain(TENANT_DOMAIN).build())); + } + + @Test + void pinsSelectorToTheTenantDomainAndEscapesFilters() { + DeviceLogFilterCriteria filter = DeviceLogFilterCriteria.builder() + .levels(List.of(DeviceLogLevel.ERROR, DeviceLogLevel.WARN)) + .search("a\"b") + .from(FROM) + .to(TO) + .build(); + + service.queryDeviceLogs(MACHINE_ID, filter, page(null, null)); + + verify(lokiClient).queryRange( + "{job=\"agent-logs\", tenant_domain=\"acme.openframe.ai\", level=~\"ERROR|WARN\"}" + + " |~ \"(?i)a\\\"b\" | machine_id=\"machine-1\"", + FROM_NANOS, TO_NANOS + 1, 21, LokiDirection.BACKWARD); + } + + @Test + void defaultsToTheLastSevenDaysWithoutLevelOrSearchFilters() { + service.queryDeviceLogs(MACHINE_ID, DeviceLogFilterCriteria.builder().to(TO).build(), page(null, null)); + + verify(lokiClient).queryRange( + "{job=\"agent-logs\", tenant_domain=\"acme.openframe.ai\"} | machine_id=\"machine-1\"", + TO_NANOS - Duration.ofDays(7).toNanos(), TO_NANOS + 1, 21, LokiDirection.BACKWARD); + } + + @Test + void rejectsDevicesNotVisibleToTheTenant() { + when(deviceService.findByMachineId("other-tenant-machine")).thenReturn(Optional.empty()); + + assertThatThrownBy(() -> service.queryDeviceLogs("other-tenant-machine", null, page(null, null))) + .isInstanceOf(DeviceNotFoundException.class); + verifyNoInteractions(lokiClient); + } + + @Test + void failsWhenTheTenantHasNoDomain() { + when(tenantRepository.findById(TENANT_ID)).thenReturn(Optional.of(Tenant.builder().id(TENANT_ID).build())); + + assertThatThrownBy(() -> service.queryDeviceLogs(MACHINE_ID, window(), page(null, null))) + .isInstanceOf(InternalException.class); + verifyNoInteractions(lokiClient); + } + + @Test + void firstPageReportsNextPageAndCursors() { + when(lokiClient.queryRange(anyString(), anyLong(), anyLong(), eq(3), eq(LokiDirection.BACKWARD))) + .thenReturn(List.of(entry(300, "c"), entry(200, "b"), entry(100, "a"))); + + GenericQueryResult result = service.queryDeviceLogs(MACHINE_ID, window(), page(2, null)); + + assertThat(result.getItems()).extracting(DeviceLogEntry::getMessage).containsExactly("c", "b"); + assertThat(result.getPageInfo().isHasNextPage()).isTrue(); + assertThat(result.getPageInfo().isHasPreviousPage()).isFalse(); + assertThat(result.getPageInfo().getStartCursor()).isEqualTo(CursorCodec.encode("300:1")); + assertThat(result.getPageInfo().getEndCursor()).isEqualTo(CursorCodec.encode("200:1")); + } + + @Test + void nextPageSkipsEntriesAlreadyReturnedAtTheCursorTimestamp() { + long cursorNanos = TO_NANOS - 500; + when(lokiClient.queryRange(anyString(), eq(FROM_NANOS), eq(cursorNanos + 1), eq(4), eq(LokiDirection.BACKWARD))) + .thenReturn(List.of(entry(cursorNanos, "b1"), entry(cursorNanos, "b2"), entry(cursorNanos - 1, "a"))); + + GenericQueryResult result = + service.queryDeviceLogs(MACHINE_ID, window(), page(2, cursorNanos + ":1")); + + assertThat(result.getItems()).extracting(DeviceLogEntry::getMessage).containsExactly("b2", "a"); + assertThat(result.getItems()).extracting(DeviceLogEntry::getCursor) + .containsExactly(CursorCodec.encode(cursorNanos + ":2"), CursorCodec.encode((cursorNanos - 1) + ":1")); + assertThat(result.getPageInfo().isHasNextPage()).isFalse(); + assertThat(result.getPageInfo().isHasPreviousPage()).isTrue(); + } + + @Test + void cursorBeforeTheWindowReturnsAnEmptyPageWithoutQuerying() { + GenericQueryResult result = + service.queryDeviceLogs(MACHINE_ID, window(), page(2, (FROM_NANOS - 1) + ":1")); + + assertThat(result.getItems()).isEmpty(); + assertThat(result.getPageInfo().isHasNextPage()).isFalse(); + verifyNoInteractions(lokiClient); + } + + @Test + void mapsStructuredMetadataOntoTheEntry() { + when(lokiClient.queryRange(anyString(), anyLong(), anyLong(), anyInt(), eq(LokiDirection.BACKWARD))) + .thenReturn(List.of(new LokiLogEntry(TO_NANOS + 35, "Control channel disconnected", Map.of( + "level", "ERROR", + "hostname", "Mishas-MacBook-Pro.local", + "agent_ts", "2026-09-14T11:59:59.487Z", + "count", "2")))); + + DeviceLogEntry entry = service.queryDeviceLogs(MACHINE_ID, window(), page(null, null)).getItems().get(0); + + assertThat(entry.getTimestamp()).isEqualTo(TO.plusNanos(35)); + assertThat(entry.getAgentTimestamp()).isEqualTo(Instant.parse("2026-09-14T11:59:59.487Z")); + assertThat(entry.getLevel()).isEqualTo("ERROR"); + assertThat(entry.getHostname()).isEqualTo("Mishas-MacBook-Pro.local"); + assertThat(entry.getCount()).isEqualTo(2L); + } + + @Test + void rejectsMalformedCursorsRangesAndSearches() { + assertThatThrownBy(() -> service.queryDeviceLogs(MACHINE_ID, window(), page(null, "garbage"))) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> service.queryDeviceLogs(MACHINE_ID, + DeviceLogFilterCriteria.builder().from(TO.minus(Duration.ofDays(31))).to(TO).build(), page(null, null))) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> service.queryDeviceLogs(MACHINE_ID, + DeviceLogFilterCriteria.builder().from(FROM).to(TO).search("x".repeat(257)).build(), page(null, null))) + .isInstanceOf(IllegalArgumentException.class); + verifyNoInteractions(lokiClient); + } + + private static DeviceLogFilterCriteria window() { + return DeviceLogFilterCriteria.builder().from(FROM).to(TO).build(); + } + + private static CursorPaginationCriteria page(Integer limit, String rawCursor) { + return CursorPaginationCriteria.builder().limit(limit).cursor(rawCursor).build(); + } + + private static LokiLogEntry entry(long timestampNanos, String line) { + return new LokiLogEntry(timestampNanos, line, Map.of("level", "INFO")); + } +} diff --git a/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/DeviceLogDataFetcher.java b/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/DeviceLogDataFetcher.java new file mode 100644 index 0000000000..9bcd56324d --- /dev/null +++ b/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/DeviceLogDataFetcher.java @@ -0,0 +1,44 @@ +package com.openframe.api.datafetcher; + +import com.netflix.graphql.dgs.DgsComponent; +import com.netflix.graphql.dgs.DgsQuery; +import com.netflix.graphql.dgs.InputArgument; +import com.openframe.api.dto.GenericConnection; +import com.openframe.api.dto.GenericEdge; +import com.openframe.api.dto.GenericQueryResult; +import com.openframe.api.dto.device.DeviceLogEntry; +import com.openframe.api.dto.device.DeviceLogFilterInput; +import com.openframe.api.mapper.GraphQLDeviceLogMapper; +import com.openframe.api.service.device.DeviceLogService; +import jakarta.validation.Valid; +import jakarta.validation.constraints.NotBlank; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.validation.annotation.Validated; + +@DgsComponent +@Slf4j +@Validated +@RequiredArgsConstructor +@ConditionalOnProperty(name = "openframe.loki.enabled", havingValue = "true") +public class DeviceLogDataFetcher { + + private final DeviceLogService deviceLogService; + private final GraphQLDeviceLogMapper mapper; + + @DgsQuery + public GenericConnection> deviceLogs( + @InputArgument @NotBlank String machineId, + @InputArgument @Valid DeviceLogFilterInput filter, + @InputArgument Integer first, + @InputArgument String after) { + + log.debug("Fetching device logs for machineId: {}, filter: {}, first: {}, after: {}", + machineId, filter, first, after); + + GenericQueryResult result = deviceLogService.queryDeviceLogs( + machineId, mapper.toFilterCriteria(filter), mapper.toCursorPaginationCriteria(first, after)); + return mapper.toConnection(result); + } +} diff --git a/openframe-api-service-core/src/main/java/com/openframe/api/dto/device/DeviceLogFilterInput.java b/openframe-api-service-core/src/main/java/com/openframe/api/dto/device/DeviceLogFilterInput.java new file mode 100644 index 0000000000..af5ee4d44d --- /dev/null +++ b/openframe-api-service-core/src/main/java/com/openframe/api/dto/device/DeviceLogFilterInput.java @@ -0,0 +1,26 @@ +package com.openframe.api.dto.device; + +import jakarta.validation.constraints.Size; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.time.Instant; +import java.util.List; + +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class DeviceLogFilterInput { + + private List levels; + + @Size(max = 256, message = "search cannot exceed 256 characters") + private String search; + + private Instant from; + + private Instant to; +} diff --git a/openframe-api-service-core/src/main/java/com/openframe/api/exception/GraphQLExceptionHandler.java b/openframe-api-service-core/src/main/java/com/openframe/api/exception/GraphQLExceptionHandler.java index b6202f8639..0dbd8aed53 100644 --- a/openframe-api-service-core/src/main/java/com/openframe/api/exception/GraphQLExceptionHandler.java +++ b/openframe-api-service-core/src/main/java/com/openframe/api/exception/GraphQLExceptionHandler.java @@ -4,6 +4,7 @@ import com.openframe.core.exception.ConflictException; import com.openframe.core.exception.ErrorCode; import com.openframe.core.exception.NotFoundException; +import com.openframe.data.loki.client.LokiQueryException; import com.openframe.data.pinot.repository.exception.PinotQueryException; import graphql.GraphQLError; import graphql.execution.DataFetcherExceptionHandlerParameters; @@ -31,6 +32,8 @@ public CompletableFuture handleException( if (exception instanceof PinotQueryException) { error = buildError("Query failed. Please try again later.", ErrorCode.PINOT_QUERY_ERROR); + } else if (exception instanceof LokiQueryException) { + error = buildError("Device logs are temporarily unavailable. Please try again later.", ErrorCode.LOKI_QUERY_ERROR); } else if (exception instanceof DataAccessException) { error = buildError("Database operation failed. Please try again later.", ErrorCode.DATABASE_ERROR); } else if (exception instanceof NotFoundException nfe) { diff --git a/openframe-api-service-core/src/main/java/com/openframe/api/mapper/GraphQLDeviceLogMapper.java b/openframe-api-service-core/src/main/java/com/openframe/api/mapper/GraphQLDeviceLogMapper.java new file mode 100644 index 0000000000..2e30677fa4 --- /dev/null +++ b/openframe-api-service-core/src/main/java/com/openframe/api/mapper/GraphQLDeviceLogMapper.java @@ -0,0 +1,56 @@ +package com.openframe.api.mapper; + +import com.openframe.api.dto.GenericConnection; +import com.openframe.api.dto.GenericEdge; +import com.openframe.api.dto.GenericQueryResult; +import com.openframe.api.dto.device.DeviceLogEntry; +import com.openframe.api.dto.device.DeviceLogFilterCriteria; +import com.openframe.api.dto.device.DeviceLogFilterInput; +import com.openframe.api.dto.shared.ConnectionArgs; +import com.openframe.api.dto.shared.CursorCodec; +import com.openframe.api.dto.shared.CursorPaginationCriteria; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +import java.util.List; + +@Component +public class GraphQLDeviceLogMapper { + + public DeviceLogFilterCriteria toFilterCriteria(DeviceLogFilterInput input) { + if (input == null) { + return DeviceLogFilterCriteria.builder().build(); + } + return DeviceLogFilterCriteria.builder() + .levels(input.getLevels()) + .search(input.getSearch()) + .from(input.getFrom()) + .to(input.getTo()) + .build(); + } + + /** + * Rejects an undecodable {@code after} instead of letting the shared helper silently restart at page one, + * which would make an infinite-scroll client loop over the newest entries. + */ + public CursorPaginationCriteria toCursorPaginationCriteria(Integer first, String after) { + if (StringUtils.hasText(after) && CursorCodec.decode(after) == null) { + throw new IllegalArgumentException("Invalid cursor"); + } + return CursorPaginationCriteria.fromConnectionArgs(ConnectionArgs.builder().first(first).after(after).build()); + } + + public GenericConnection> toConnection(GenericQueryResult result) { + List> edges = result.getItems().stream() + .map(entry -> GenericEdge.builder() + .node(entry) + .cursor(entry.getCursor()) + .build()) + .toList(); + + return GenericConnection.>builder() + .edges(edges) + .pageInfo(result.getPageInfo()) + .build(); + } +} diff --git a/openframe-api-service-core/src/main/resources/schema/device-log.graphqls b/openframe-api-service-core/src/main/resources/schema/device-log.graphqls new file mode 100644 index 0000000000..8fd6160139 --- /dev/null +++ b/openframe-api-service-core/src/main/resources/schema/device-log.graphqls @@ -0,0 +1,43 @@ +extend type Query { + # Agent logs of one device, newest first, always scoped to the current tenant. + # Paginate with first/after (older entries); re-query without `after` to pick up newer ones. + deviceLogs( + machineId: String! + filter: DeviceLogFilterInput + first: Int + after: String + ): DeviceLogConnection! +} + +enum DeviceLogLevel { + DEBUG + INFO + WARN + ERROR +} + +input DeviceLogFilterInput { + levels: [DeviceLogLevel!] # Any of these levels; all levels when omitted + search: String # Case-insensitive substring match on the message, max 256 characters + from: Instant # Inclusive; defaults to 7 days before `to` + to: Instant # Inclusive; defaults to now. The range may not exceed 30 days +} + +type DeviceLogConnection { + edges: [DeviceLogEdge!]! + pageInfo: PageInfo! +} + +type DeviceLogEdge { + node: DeviceLogEntry! + cursor: String! +} + +type DeviceLogEntry { + timestamp: Instant! # Ingestion time with nanosecond precision; the sort key + agentTimestamp: Instant # When the agent wrote the line + level: String! + message: String! + hostname: String + count: Long # Identical lines the agent collapsed into this entry +} diff --git a/openframe-api-service-core/src/test/java/com/openframe/api/mapper/GraphQLDeviceLogMapperTest.java b/openframe-api-service-core/src/test/java/com/openframe/api/mapper/GraphQLDeviceLogMapperTest.java new file mode 100644 index 0000000000..a0ade3df61 --- /dev/null +++ b/openframe-api-service-core/src/test/java/com/openframe/api/mapper/GraphQLDeviceLogMapperTest.java @@ -0,0 +1,28 @@ +package com.openframe.api.mapper; + +import com.openframe.api.dto.shared.CursorCodec; +import com.openframe.api.dto.shared.CursorPaginationCriteria; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class GraphQLDeviceLogMapperTest { + + private final GraphQLDeviceLogMapper mapper = new GraphQLDeviceLogMapper(); + + @Test + void decodesTheAfterCursor() { + CursorPaginationCriteria criteria = mapper.toCursorPaginationCriteria(50, CursorCodec.encode("123:1")); + + assertThat(criteria.getCursor()).isEqualTo("123:1"); + assertThat(criteria.getLimit()).isEqualTo(50); + assertThat(criteria.isBackward()).isFalse(); + } + + @Test + void rejectsAnUndecodableCursorInsteadOfRestartingAtPageOne() { + assertThatThrownBy(() -> mapper.toCursorPaginationCriteria(20, "!!not-base64!!")) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/openframe-data-loki/pom.xml b/openframe-data-loki/pom.xml new file mode 100644 index 0000000000..0c59634269 --- /dev/null +++ b/openframe-data-loki/pom.xml @@ -0,0 +1,39 @@ + + + 4.0.0 + + + com.openframe.oss + openframe-oss-lib + ${revision} + + + openframe-data-loki + jar + OpenFrame Data Loki + Grafana Loki query client and configuration for OpenFrame platform + + + + org.springframework.boot + spring-boot-starter + + + + org.springframework + spring-web + + + com.fasterxml.jackson.core + jackson-databind + + + + org.springframework.boot + spring-boot-starter-test + test + + + diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LogQl.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LogQl.java new file mode 100644 index 0000000000..3b3f1f04cc --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LogQl.java @@ -0,0 +1,52 @@ +package com.openframe.data.loki.client; + +/** + * Builds LogQL literals from untrusted values. Every value placed into a query must go through here, + * otherwise a quote in user input can widen the stream selector to other tenants' logs. + */ +public final class LogQl { + + private static final String REGEX_METACHARACTERS = "\\.+*?()|[]{}^$"; + + private LogQl() { + } + + /** + * A double-quoted LogQL string literal holding {@code value} verbatim (Go string escaping). + */ + public static String quote(String value) { + StringBuilder quoted = new StringBuilder(value.length() + 2).append('"'); + for (char c : value.toCharArray()) { + switch (c) { + case '"' -> quoted.append("\\\""); + case '\\' -> quoted.append("\\\\"); + case '\n' -> quoted.append("\\n"); + case '\r' -> quoted.append("\\r"); + case '\t' -> quoted.append("\\t"); + default -> { + if (c < 0x20 || c == 0x7f) { + quoted.append(String.format("\\u%04x", (int) c)); + } else { + quoted.append(c); + } + } + } + } + return quoted.append('"').toString(); + } + + /** + * An RE2 pattern matching {@code value} literally; mirrors Go's {@code regexp.QuoteMeta}. + * The result is a pattern, not a literal: wrap it with {@link #quote(String)} before use. + */ + public static String regexLiteral(String value) { + StringBuilder escaped = new StringBuilder(value.length() * 2); + for (char c : value.toCharArray()) { + if (REGEX_METACHARACTERS.indexOf(c) >= 0) { + escaped.append('\\'); + } + escaped.append(c); + } + return escaped.toString(); + } +} diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiClient.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiClient.java new file mode 100644 index 0000000000..9014ec56bd --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiClient.java @@ -0,0 +1,90 @@ +package com.openframe.data.loki.client; + +import com.openframe.data.loki.model.LokiDirection; +import com.openframe.data.loki.model.LokiLogEntry; +import com.openframe.data.loki.model.LokiQueryResponse; +import org.springframework.web.client.RestClient; +import org.springframework.web.client.RestClientException; +import org.springframework.web.client.RestClientResponseException; +import org.springframework.web.client.support.RestClientAdapter; +import org.springframework.web.service.invoker.HttpServiceProxyFactory; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.SortedMap; +import java.util.TreeMap; + +public class LokiClient { + + private static final String STREAMS_RESULT_TYPE = "streams"; + private static final int MAX_ERROR_BODY_LENGTH = 500; + + private final LokiHttpApi api; + + public LokiClient(RestClient restClient) { + this.api = HttpServiceProxyFactory.builderFor(RestClientAdapter.create(restClient)) + .build() + .createClient(LokiHttpApi.class); + } + + /** + * Runs a LogQL log query over {@code [startNanos, endNanos)} and returns up to {@code limit} entries merged + * across streams: newest first for {@link LokiDirection#BACKWARD}, oldest first for + * {@link LokiDirection#FORWARD}. Entries sharing a timestamp are ordered by labels and line, so repeating + * a query returns them in the same order. + */ + public List queryRange(String query, long startNanos, long endNanos, int limit, + LokiDirection direction) { + LokiQueryResponse response; + try { + response = api.queryRange(query, startNanos, endNanos, limit, direction.name().toLowerCase(Locale.ROOT)); + } catch (RestClientResponseException e) { + throw new LokiQueryException("Loki query failed with HTTP " + e.getStatusCode().value() + ": " + + abbreviate(e.getResponseBodyAsString()), e); + } catch (RestClientException e) { + throw new LokiQueryException("Loki query failed: " + e.getMessage(), e); + } + return toEntries(response, direction); + } + + private static List toEntries(LokiQueryResponse response, LokiDirection direction) { + if (response == null || response.data() == null || response.data().result() == null) { + return List.of(); + } + if (!STREAMS_RESULT_TYPE.equals(response.data().resultType())) { + throw new LokiQueryException("Expected a log query result, got: " + response.data().resultType()); + } + + List entries = new ArrayList<>(); + for (LokiQueryResponse.LogStream stream : response.data().result()) { + if (stream.values() == null) { + continue; + } + // Sorted, so equal-timestamp ordering below does not depend on JSON key order + SortedMap labels = Collections.unmodifiableSortedMap( + new TreeMap<>(stream.stream() != null ? stream.stream() : Map.of())); + for (List value : stream.values()) { + if (value != null && value.size() >= 2) { + entries.add(new LokiLogEntry(Long.parseLong(value.get(0)), value.get(1), labels)); + } + } + } + + Comparator byTimestamp = Comparator.comparingLong(LokiLogEntry::timestampNanos); + entries.sort((direction == LokiDirection.BACKWARD ? byTimestamp.reversed() : byTimestamp) + .thenComparing(entry -> entry.labels().toString()) + .thenComparing(LokiLogEntry::line)); + return entries; + } + + private static String abbreviate(String body) { + if (body == null) { + return ""; + } + return body.length() <= MAX_ERROR_BODY_LENGTH ? body : body.substring(0, MAX_ERROR_BODY_LENGTH) + "..."; + } +} diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiHttpApi.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiHttpApi.java new file mode 100644 index 0000000000..9880c7ce70 --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiHttpApi.java @@ -0,0 +1,24 @@ +package com.openframe.data.loki.client; + +import com.openframe.data.loki.model.LokiQueryResponse; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.service.annotation.GetExchange; +import org.springframework.web.service.annotation.HttpExchange; + +/** + * Declarative binding of the Loki HTTP query API, limited to what OpenFrame reads. + * Request parameters are sent as encoded URI variables, so LogQL braces, pipes and {@code +} survive intact. + */ +@HttpExchange("/loki/api/v1") +public interface LokiHttpApi { + + /** + * Timestamps are nanosecond Unix epochs: {@code start} is inclusive, {@code end} exclusive. + */ + @GetExchange("/query_range") + LokiQueryResponse queryRange(@RequestParam("query") String query, + @RequestParam("start") long startNanos, + @RequestParam("end") long endNanos, + @RequestParam("limit") int limit, + @RequestParam("direction") String direction); +} diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiQueryException.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiQueryException.java new file mode 100644 index 0000000000..91d805645e --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/client/LokiQueryException.java @@ -0,0 +1,12 @@ +package com.openframe.data.loki.client; + +public class LokiQueryException extends RuntimeException { + + public LokiQueryException(String message) { + super(message); + } + + public LokiQueryException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/config/LokiConfig.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/config/LokiConfig.java new file mode 100644 index 0000000000..e135553961 --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/config/LokiConfig.java @@ -0,0 +1,37 @@ +package com.openframe.data.loki.config; + +import com.openframe.data.loki.client.LokiClient; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.web.client.ClientHttpRequestFactories; +import org.springframework.boot.web.client.ClientHttpRequestFactorySettings; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.util.Assert; +import org.springframework.web.client.RestClient; + +@Configuration +@ConditionalOnProperty(name = "openframe.loki.enabled", havingValue = "true") +@EnableConfigurationProperties(LokiProperties.class) +public class LokiConfig { + + /** + * Built on Boot's auto-configured {@link RestClient.Builder} when there is one, so Loki calls carry the + * standard {@code http.client.requests} observation (metrics and tracing) like other outbound calls. + */ + @Bean + public LokiClient lokiClient(ObjectProvider restClientBuilder, LokiProperties properties) { + Assert.hasText(properties.getUrl(), "openframe.loki.url must be set when openframe.loki.enabled=true"); + + ClientHttpRequestFactorySettings settings = ClientHttpRequestFactorySettings.DEFAULTS + .withConnectTimeout(properties.getConnectTimeout()) + .withReadTimeout(properties.getReadTimeout()); + + RestClient restClient = restClientBuilder.getIfAvailable(RestClient::builder) + .baseUrl(properties.getUrl()) + .requestFactory(ClientHttpRequestFactories.get(settings)) + .build(); + return new LokiClient(restClient); + } +} diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/config/LokiProperties.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/config/LokiProperties.java new file mode 100644 index 0000000000..89bee43cc0 --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/config/LokiProperties.java @@ -0,0 +1,31 @@ +package com.openframe.data.loki.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +import java.time.Duration; + +/** + * Loki query connection settings. + *

+ * Lives under {@code openframe.loki} rather than {@code loki}: OSS deployments already use + * {@code loki.url} for the logback push appender. + */ +@Data +@ConfigurationProperties(prefix = "openframe.loki") +public class LokiProperties { + + private boolean enabled; + + /** + * Base URL of the Loki gateway, e.g. {@code http://loki.internal.openframe.ai:80}. + */ + private String url; + + private Duration connectTimeout = Duration.ofSeconds(2); + + /** + * Upper bound for one query. These are user-facing reads, so it stays below Loki's own query timeout. + */ + private Duration readTimeout = Duration.ofSeconds(30); +} diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiDirection.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiDirection.java new file mode 100644 index 0000000000..364f91353d --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiDirection.java @@ -0,0 +1,6 @@ +package com.openframe.data.loki.model; + +public enum LokiDirection { + FORWARD, + BACKWARD +} diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiLogEntry.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiLogEntry.java new file mode 100644 index 0000000000..4636b38187 --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiLogEntry.java @@ -0,0 +1,14 @@ +package com.openframe.data.loki.model; + +import java.time.Instant; +import java.util.Map; + +/** + * A log line with its stream labels and structured metadata merged into {@code labels}. + */ +public record LokiLogEntry(long timestampNanos, String line, Map labels) { + + public Instant timestamp() { + return Instant.ofEpochSecond(0, timestampNanos); + } +} diff --git a/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiQueryResponse.java b/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiQueryResponse.java new file mode 100644 index 0000000000..fb601d0fb4 --- /dev/null +++ b/openframe-data-loki/src/main/java/com/openframe/data/loki/model/LokiQueryResponse.java @@ -0,0 +1,23 @@ +package com.openframe.data.loki.model; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; + +import java.util.List; +import java.util.Map; + +@JsonIgnoreProperties(ignoreUnknown = true) +public record LokiQueryResponse(String status, QueryData data) { + + @JsonIgnoreProperties(ignoreUnknown = true) + public record QueryData(String resultType, List result) { + } + + /** + * One log stream. Without the categorize-labels encoding flag Loki folds structured metadata into + * {@code stream}, so each distinct metadata set arrives as its own stream. Each value is a + * {@code [timestampNanos, line]} pair. + */ + @JsonIgnoreProperties(ignoreUnknown = true) + public record LogStream(Map stream, List> values) { + } +} diff --git a/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LogQlTest.java b/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LogQlTest.java new file mode 100644 index 0000000000..e9e59eeabb --- /dev/null +++ b/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LogQlTest.java @@ -0,0 +1,24 @@ +package com.openframe.data.loki.client; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class LogQlTest { + + @Test + void quoteEscapesQuotesBackslashesAndControlCharacters() { + assertThat(LogQl.quote("a\"b\\c\nd")).isEqualTo("\"a\\\"b\\\\c\\nd\\u0001\""); + } + + @Test + void quoteKeepsHostileInputInsideTheLiteral() { + assertThat(LogQl.quote("x\"} or {job=~\".+")).isEqualTo("\"x\\\"} or {job=~\\\".+\""); + } + + @Test + void regexLiteralEscapesRe2Metacharacters() { + assertThat(LogQl.regexLiteral("a.b*(c)|[d]{e}^$\\+?")) + .isEqualTo("a\\.b\\*\\(c\\)\\|\\[d\\]\\{e\\}\\^\\$\\\\\\+\\?"); + } +} diff --git a/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LokiClientTest.java b/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LokiClientTest.java new file mode 100644 index 0000000000..941f79c59b --- /dev/null +++ b/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LokiClientTest.java @@ -0,0 +1,97 @@ +package com.openframe.data.loki.client; + +import com.openframe.data.loki.model.LokiDirection; +import com.openframe.data.loki.model.LokiLogEntry; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpMethod; +import org.springframework.http.MediaType; +import org.springframework.test.web.client.MockRestServiceServer; +import org.springframework.web.client.RestClient; + +import java.net.URI; +import java.net.URLDecoder; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.hamcrest.Matchers.startsWith; +import static org.springframework.test.web.client.match.MockRestRequestMatchers.method; +import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo; +import static org.springframework.test.web.client.response.MockRestResponseCreators.withBadRequest; +import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess; + +class LokiClientTest { + + private static final String EMPTY_RESPONSE = + "{\"status\":\"success\",\"data\":{\"resultType\":\"streams\",\"result\":[]}}"; + + private MockRestServiceServer server; + private LokiClient client; + + @BeforeEach + void setUp() { + RestClient.Builder builder = RestClient.builder().baseUrl("http://loki.test"); + server = MockRestServiceServer.bindTo(builder).build(); + client = new LokiClient(builder.build()); + } + + @Test + void sendsQueryRangeParametersEncoded() { + // '+' would reach Loki as a space if left unencoded + String query = "{job=\"agent-logs\"} |~ \"(?i)a+b\" | machine_id=\"m-1\""; + server.expect(requestTo(startsWith("http://loki.test/loki/api/v1/query_range?"))) + .andExpect(method(HttpMethod.GET)) + .andExpect(request -> assertThat(queryParams(request.getURI())) + .containsEntry("query", query) + .containsEntry("start", "100") + .containsEntry("end", "200") + .containsEntry("limit", "5") + .containsEntry("direction", "backward")) + .andRespond(withSuccess(EMPTY_RESPONSE, MediaType.APPLICATION_JSON)); + + assertThat(client.queryRange(query, 100, 200, 5, LokiDirection.BACKWARD)).isEmpty(); + server.verify(); + } + + @Test + void mergesStreamsNewestFirstForBackwardQueries() { + server.expect(requestTo(startsWith("http://loki.test/loki/api/v1/query_range"))) + .andRespond(withSuccess(""" + {"status":"success","data":{"resultType":"streams","result":[ + {"stream":{"level":"INFO","machine_id":"m-1"},"values":[["300","c"],["100","a"]]}, + {"stream":{"level":"ERROR","machine_id":"m-1"},"values":[["200","b"]]} + ],"stats":{}}} + """, MediaType.APPLICATION_JSON)); + + List entries = client.queryRange("{job=\"x\"}", 0, 400, 10, LokiDirection.BACKWARD); + + assertThat(entries).extracting(LokiLogEntry::line).containsExactly("c", "b", "a"); + assertThat(entries.get(1).labels()).containsEntry("level", "ERROR"); + assertThat(entries.get(0).timestamp().getNano()).isEqualTo(300); + } + + @Test + void wrapsLokiErrorsWithStatusAndBody() { + server.expect(requestTo(startsWith("http://loki.test/"))) + .andRespond(withBadRequest().body("parse error at line 1").contentType(MediaType.TEXT_PLAIN)); + + assertThatThrownBy(() -> client.queryRange("{", 0, 1, 1, LokiDirection.BACKWARD)) + .isInstanceOf(LokiQueryException.class) + .hasMessageContaining("HTTP 400") + .hasMessageContaining("parse error"); + } + + private static Map queryParams(URI uri) { + Map params = new HashMap<>(); + for (String pair : uri.getRawQuery().split("&")) { + int separator = pair.indexOf('='); + params.put(URLDecoder.decode(pair.substring(0, separator), UTF_8), + URLDecoder.decode(pair.substring(separator + 1), UTF_8)); + } + return params; + } +} diff --git a/openframe-exception/src/main/java/com/openframe/core/exception/ErrorCode.java b/openframe-exception/src/main/java/com/openframe/core/exception/ErrorCode.java index da5cf744ae..9145163a9a 100644 --- a/openframe-exception/src/main/java/com/openframe/core/exception/ErrorCode.java +++ b/openframe-exception/src/main/java/com/openframe/core/exception/ErrorCode.java @@ -66,6 +66,7 @@ public enum ErrorCode { // Infrastructure error codes TYPE_MISMATCH("TYPE_MISMATCH", 400), PINOT_QUERY_ERROR("PINOT_QUERY_ERROR", 503), + LOKI_QUERY_ERROR("LOKI_QUERY_ERROR", 503), DATABASE_ERROR("DATABASE_ERROR", 503); private final String code; diff --git a/pom.xml b/pom.xml index 8226a9d8f1..0da1c98086 100644 --- a/pom.xml +++ b/pom.xml @@ -53,6 +53,7 @@ openframe-api-lib openframe-data-cassandra openframe-data-pinot + openframe-data-loki openframe-data-nats openframe-data-device-aspect openframe-data-timeentry-aspect @@ -162,6 +163,11 @@ openframe-data-pinot ${revision} + + com.openframe.oss + openframe-data-loki + ${revision} + com.openframe.oss openframe-data-nats From fe9b44f4cd180b67d1d6d8c5fe89b183afb09db4 Mon Sep 17 00:00:00 2001 From: Aliaska Varieva Date: Mon, 14 Sep 2026 21:22:14 +0200 Subject: [PATCH 2/3] perf(devices): cache the tenant domain for device logs A tenant pod serves one tenant and tenant domains never change, so resolve the domain from Mongo once per pod instead of on every deviceLogs request. A missing domain is not cached, so a tenant still being provisioned recovers on the next call. --- .../api/service/device/DeviceLogService.java | 21 +++++++++++++--- .../service/device/DeviceLogServiceTest.java | 25 +++++++++++++++++++ 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java index c42dfd21af..35568141c5 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java @@ -26,6 +26,8 @@ import java.time.format.DateTimeParseException; import java.util.ArrayList; import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import static java.util.stream.Collectors.joining; @@ -51,6 +53,7 @@ public class DeviceLogService { private final DeviceService deviceService; private final TenantIdProvider tenantIdProvider; private final TenantRepository tenantRepository; + private final Map tenantDomains = new ConcurrentHashMap<>(); /** * Logs of one device, newest first. @@ -114,15 +117,25 @@ static String buildQuery(String tenantDomain, String machineId, DeviceLogFilterC return query.append(" | machine_id=").append(LogQl.quote(machineId)).toString(); } + /** + * Cached for the life of the pod: a tenant pod serves one tenant and tenant domains never change. A missing + * domain is not cached, so a tenant that is still being provisioned recovers on the next call. + */ private String resolveTenantDomain() { String tenantId = tenantIdProvider.getTenantId(); + String domain = tenantDomains.computeIfAbsent(tenantId, this::findTenantDomain); + if (domain == null) { + log.error("Cannot query device logs: tenant {} has no domain", tenantId); + throw new InternalException("Device logs are not available for this tenant"); + } + return domain; + } + + private String findTenantDomain(String tenantId) { return tenantRepository.findById(tenantId) .map(Tenant::getDomain) .filter(StringUtils::hasText) - .orElseThrow(() -> { - log.error("Cannot query device logs: tenant {} has no domain", tenantId); - return new InternalException("Device logs are not available for this tenant"); - }); + .orElse(null); } /** diff --git a/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java b/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java index d88e817567..95878157b7 100644 --- a/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java +++ b/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java @@ -35,7 +35,9 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.startsWith; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; @@ -112,6 +114,29 @@ void failsWhenTheTenantHasNoDomain() { verifyNoInteractions(lokiClient); } + @Test + void cachesTheTenantDomainAcrossRequests() { + service.queryDeviceLogs(MACHINE_ID, window(), page(null, null)); + service.queryDeviceLogs(MACHINE_ID, window(), page(null, null)); + + verify(tenantRepository, times(1)).findById(TENANT_ID); + } + + @Test + void doesNotCacheAMissingDomain() { + when(tenantRepository.findById(TENANT_ID)) + .thenReturn(Optional.of(Tenant.builder().id(TENANT_ID).build())) + .thenReturn(Optional.of(Tenant.builder().id(TENANT_ID).domain(TENANT_DOMAIN).build())); + + assertThatThrownBy(() -> service.queryDeviceLogs(MACHINE_ID, window(), page(null, null))) + .isInstanceOf(InternalException.class); + service.queryDeviceLogs(MACHINE_ID, window(), page(null, null)); + + verify(tenantRepository, times(2)).findById(TENANT_ID); + verify(lokiClient).queryRange(startsWith("{job=\"agent-logs\", tenant_domain=\"acme.openframe.ai\""), + anyLong(), anyLong(), anyInt(), eq(LokiDirection.BACKWARD)); + } + @Test void firstPageReportsNextPageAndCursors() { when(lokiClient.queryRange(anyString(), anyLong(), anyLong(), eq(3), eq(LokiDirection.BACKWARD))) From f75a87875b254f36d8169bc46a108c2bc5bcb934 Mon Sep 17 00:00:00 2001 From: Aliaska Varieva Date: Tue, 15 Sep 2026 03:00:29 +0200 Subject: [PATCH 3/3] feat(devices): 500-line device log pages, whole-timestamp cursor, real-Loki tests Serve up to 500 lines per deviceLogs page (default 100): the agent ships up to 50 lines a minute per device, so the shared 100-item cap covered only a couple of minutes of a busy device. Pages no longer split lines that share a timestamp. Loki cuts a result at the limit without regard to timestamps, so the previous skip-based cursor could drop or repeat lines at a shared timestamp. The cursor is now the last line's timestamp and the next page ends exactly at it. Add LokiClientIT and DeviceLogServiceIT, which run against Loki 3.7.3 in Testcontainers during the normal test phase. --- openframe-api-lib/pom.xml | 5 + .../api/service/device/DeviceLogCursor.java | 21 +- .../api/service/device/DeviceLogService.java | 76 +++-- .../service/device/DeviceLogServiceIT.java | 280 ++++++++++++++++++ .../service/device/DeviceLogServiceTest.java | 58 +++- .../main/resources/schema/device-log.graphqls | 2 +- openframe-data-loki/pom.xml | 5 + .../data/loki/client/LokiClientIT.java | 134 +++++++++ 8 files changed, 523 insertions(+), 58 deletions(-) create mode 100644 openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceIT.java create mode 100644 openframe-data-loki/src/test/java/com/openframe/data/loki/client/LokiClientIT.java diff --git a/openframe-api-lib/pom.xml b/openframe-api-lib/pom.xml index e435d4e41a..acc0dcd142 100644 --- a/openframe-api-lib/pom.xml +++ b/openframe-api-lib/pom.xml @@ -99,5 +99,10 @@ spring-boot-starter-test test + + org.testcontainers + testcontainers + test + \ No newline at end of file diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogCursor.java b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogCursor.java index f69277bf17..29d195ba89 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogCursor.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogCursor.java @@ -3,16 +3,13 @@ import com.openframe.api.dto.shared.CursorCodec; /** - * Position in a newest-first device log listing: the Loki timestamp of the last returned entry and how many - * entries at exactly that timestamp have been returned so far. Loki has no offsets, so the next page is read - * up to and including that timestamp and the already returned entries at it are skipped. + * Position in a newest-first device log listing: the Loki timestamp of the last returned line. A page never splits + * the lines that share a timestamp, so the next page is read from strictly before it. */ -record DeviceLogCursor(long timestampNanos, int skip) { - - static final int MAX_SKIP = 1000; +record DeviceLogCursor(long timestampNanos) { String encode() { - return CursorCodec.encode(timestampNanos + ":" + skip); + return CursorCodec.encode(String.valueOf(timestampNanos)); } /** @@ -23,13 +20,11 @@ static DeviceLogCursor fromRaw(String raw) { return null; } try { - int separator = raw.indexOf(':'); - long timestampNanos = Long.parseLong(raw.substring(0, separator)); - int skip = Integer.parseInt(raw.substring(separator + 1)); - if (timestampNanos > 0 && skip >= 0 && skip <= MAX_SKIP) { - return new DeviceLogCursor(timestampNanos, skip); + long timestampNanos = Long.parseLong(raw); + if (timestampNanos > 0) { + return new DeviceLogCursor(timestampNanos); } - } catch (RuntimeException ignored) { + } catch (NumberFormatException ignored) { // fall through to the rejection below } throw new IllegalArgumentException("Invalid cursor"); diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java index 35568141c5..4725083aa1 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/device/DeviceLogService.java @@ -45,6 +45,10 @@ public class DeviceLogService { static final Duration DEFAULT_LOOKBACK = Duration.ofDays(7); static final Duration MAX_RANGE = Duration.ofDays(30); static final int MAX_SEARCH_LENGTH = 256; + static final int DEFAULT_PAGE_SIZE = 100; + static final int MAX_PAGE_SIZE = 500; + // Loki's max_entries_limit_per_query on prod + static final int MAX_LINES_PER_TIMESTAMP = 5000; private static final String AGENT_LOGS_JOB = "agent-logs"; private static final long NANOS_PER_SECOND = 1_000_000_000L; @@ -68,7 +72,7 @@ public GenericQueryResult queryDeviceLogs(String machineId, .orElseThrow(() -> new DeviceNotFoundException("Machine not found: " + machineId)); DeviceLogFilterCriteria criteria = filter != null ? filter : new DeviceLogFilterCriteria(); - CursorPaginationCriteria page = (pagination != null ? pagination : new CursorPaginationCriteria()).normalize(); + CursorPaginationCriteria page = pagination != null ? pagination : new CursorPaginationCriteria(); DeviceLogCursor after = DeviceLogCursor.fromRaw(page.getCursor()); validateSearch(criteria.getSearch()); @@ -77,28 +81,27 @@ public GenericQueryResult queryDeviceLogs(String machineId, validateRange(from, to); long startNanos = toNanos(from); - // Loki's end is exclusive while both the filter's upper bound and the cursor entry are inclusive + // Loki's end is exclusive: 1 ns past the inclusive upper bound, or the cursor's timestamp itself, whose lines + // the previous page returned in full long endNanos = toNanos(to) + 1; - int skip = 0; if (after != null) { - if (after.timestampNanos() < startNanos) { + if (after.timestampNanos() <= startNanos) { return result(List.of(), false, true); } - endNanos = Math.min(endNanos, after.timestampNanos() + 1); - skip = after.skip(); + endNanos = Math.min(endNanos, after.timestampNanos()); } String query = buildQuery(resolveTenantDomain(), machineId, criteria); - int pageSize = page.getLimit(); + int pageSize = pageSize(page.getLimit()); log.debug("Querying device logs for machineId: {}, query: {}, start: {}, end: {}", machineId, query, startNanos, endNanos); - // One extra entry tells whether there is a next page - List entries = lokiClient.queryRange(query, startNanos, endNanos, pageSize + skip + 1, - LokiDirection.BACKWARD); - List remaining = dropReturned(entries, after); + // One extra line tells whether there is a next page + int queryLimit = pageSize + 1; + List entries = lokiClient.queryRange(query, startNanos, endNanos, queryLimit, LokiDirection.BACKWARD); + List pageEntries = wholeTimestampsOnly(entries, pageSize, query); + List items = toItems(pageEntries); - return result(toItems(remaining.subList(0, Math.min(pageSize, remaining.size())), after), - remaining.size() > pageSize, after != null); + return result(items, entries.size() > pageSize, after != null); } static String buildQuery(String tenantDomain, String machineId, DeviceLogFilterCriteria criteria) { @@ -139,30 +142,28 @@ private String findTenantDomain(String tenantId) { } /** - * Drops the entries at the cursor's timestamp that earlier pages already returned; they come first - * because the query ends right after that timestamp. + * Loki cuts a result at the limit without regard to timestamps, and does not guarantee which of the lines sharing + * the cut timestamp it keeps. The page therefore ends before that timestamp, and the next page starts with all of + * its lines. When one timestamp fills the whole page, its lines are fetched in full instead. */ - private static List dropReturned(List entries, DeviceLogCursor after) { - int index = 0; - if (after != null) { - while (index < entries.size() && index < after.skip() - && entries.get(index).timestampNanos() == after.timestampNanos()) { - index++; - } + private List wholeTimestampsOnly(List entries, int pageSize, String query) { + if (entries.size() <= pageSize) { + return entries; + } + long cutNanos = entries.get(pageSize).timestampNanos(); + int end = pageSize; + while (end > 0 && entries.get(end - 1).timestampNanos() == cutNanos) { + end--; } - return entries.subList(index, entries.size()); + if (end > 0) { + return entries.subList(0, end); + } + return lokiClient.queryRange(query, cutNanos, cutNanos + 1, MAX_LINES_PER_TIMESTAMP, LokiDirection.BACKWARD); } - private static List toItems(List entries, DeviceLogCursor after) { + private static List toItems(List entries) { List items = new ArrayList<>(entries.size()); - long runTimestamp = after != null ? after.timestampNanos() : Long.MIN_VALUE; - int runLength = after != null ? after.skip() : 0; for (LokiLogEntry entry : entries) { - if (entry.timestampNanos() != runTimestamp) { - runTimestamp = entry.timestampNanos(); - runLength = 0; - } - runLength++; items.add(DeviceLogEntry.builder() .timestamp(entry.timestamp()) .agentTimestamp(parseInstant(entry.labels().get("agent_ts"))) @@ -170,7 +171,7 @@ private static List toItems(List entries, DeviceLo .message(entry.line()) .hostname(entry.labels().get("hostname")) .count(parseLong(entry.labels().get("count"))) - .cursor(new DeviceLogCursor(runTimestamp, runLength).encode()) + .cursor(new DeviceLogCursor(entry.timestampNanos()).encode()) .build()); } return items; @@ -189,6 +190,17 @@ private static GenericQueryResult result(List it .build(); } + /** + * Larger than the shared 100-item cap: the agent ships up to 50 lines a minute per device, so 100 lines cover + * only a couple of minutes of a busy device. + */ + private static int pageSize(Integer requested) { + if (requested == null) { + return DEFAULT_PAGE_SIZE; + } + return Math.min(Math.max(requested, 1), MAX_PAGE_SIZE); + } + private static void validateSearch(String search) { if (search != null && search.length() > MAX_SEARCH_LENGTH) { throw new IllegalArgumentException("search cannot exceed " + MAX_SEARCH_LENGTH + " characters"); diff --git a/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceIT.java b/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceIT.java new file mode 100644 index 0000000000..43d117ca3c --- /dev/null +++ b/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceIT.java @@ -0,0 +1,280 @@ +package com.openframe.api.service.device; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.openframe.api.dto.GenericQueryResult; +import com.openframe.api.dto.device.DeviceLogEntry; +import com.openframe.api.dto.device.DeviceLogFilterCriteria; +import com.openframe.api.dto.device.DeviceLogLevel; +import com.openframe.api.dto.shared.CursorCodec; +import com.openframe.api.dto.shared.CursorPaginationCriteria; +import com.openframe.data.document.device.Machine; +import com.openframe.data.document.tenant.Tenant; +import com.openframe.data.loki.client.LokiClient; +import com.openframe.data.loki.model.LokiDirection; +import com.openframe.data.repository.tenant.TenantRepository; +import com.openframe.data.service.TenantIdProvider; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.springframework.web.client.RestClient; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.utility.DockerImageName; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * {@link DeviceLogService} with its real {@link LokiClient} against a real Loki, seeded the way + * {@code openframe-saas-logs-stream} writes agent logs: {@code job}, {@code tenant_domain} and {@code level} stream + * labels, with {@code machine_id}, {@code hostname}, {@code agent_ts} and {@code count} as structured metadata. + * Only the Mongo lookups (device, tenant) are stubbed. + */ +class DeviceLogServiceIT { + + private static final GenericContainer LOKI = new GenericContainer<>(DockerImageName.parse("grafana/loki:3.7.3")) + .withExposedPorts(3100) + .waitingFor(Wait.forHttp("/ready").forPort(3100).forStatusCode(200) + .withStartupTimeout(Duration.ofMinutes(2))); + + private static final String TENANT_ID = "tenant-a"; + private static final String TENANT_DOMAIN = "acme.openframe.test"; + private static final String OTHER_TENANT_DOMAIN = "globex.openframe.test"; + private static final String MACHINE_ID = "machine-a"; + private static final String OTHER_MACHINE_ID = "machine-b"; + private static final String POLL_MACHINE_ID = "machine-poll"; + + private static final int DEVICE_LINES = 600; + private static final int ERROR_LINES = 12; + private static final int TIED_LINES = 5; + private static final int TIED_ERROR_LINES = 2; + private static final long STEP_NANOS = 100_000_000L; + private static final Instant BASE = Instant.now().minus(Duration.ofHours(1)).truncatedTo(ChronoUnit.MILLIS); + private static final long BASE_NANOS = toNanos(BASE); + // Five lines share this nanosecond, between line 300 and line 301, split across the INFO and ERROR streams + private static final long TIED_NANOS = BASE_NANOS + 300 * STEP_NANOS + 50; + private static final Instant FROM = BASE.minus(Duration.ofMinutes(1)); + + private static final ObjectMapper JSON = new ObjectMapper(); + + private static LokiClient lokiClient; + private static DeviceLogService service; + + @BeforeAll + static void startLokiAndSeedLogs() throws Exception { + LOKI.start(); + lokiClient = new LokiClient(RestClient.builder().baseUrl(baseUrl()).build()); + service = newService(); + + Map>> deviceLinesByLevel = new LinkedHashMap<>(); + for (int i = 0; i < DEVICE_LINES; i++) { + String level = i % 50 == 0 ? "ERROR" : i % 20 == 0 ? "WARN" : "INFO"; + String message = String.format("line-%03d %s", i, "ERROR".equals(level) ? "Connection FAILED" : "heartbeat ok"); + deviceLinesByLevel.computeIfAbsent(level, key -> new ArrayList<>()) + .add(entry(BASE_NANOS + i * STEP_NANOS, message, MACHINE_ID)); + } + for (int i = 0; i < TIED_LINES; i++) { + String level = i < TIED_LINES - TIED_ERROR_LINES ? "INFO" : "ERROR"; + deviceLinesByLevel.computeIfAbsent(level, key -> new ArrayList<>()) + .add(entry(TIED_NANOS, "tied-" + i, MACHINE_ID)); + } + for (Map.Entry>> stream : deviceLinesByLevel.entrySet()) { + push(TENANT_DOMAIN, stream.getKey(), stream.getValue()); + } + + List> otherMachineLines = new ArrayList<>(); + List> otherTenantLines = new ArrayList<>(); + for (int i = 0; i < 20; i++) { + otherMachineLines.add(entry(BASE_NANOS + i * STEP_NANOS + 7, String.format("other-machine-%02d", i), OTHER_MACHINE_ID)); + otherTenantLines.add(entry(BASE_NANOS + i * STEP_NANOS + 9, String.format("other-tenant-%02d", i), MACHINE_ID)); + } + push(TENANT_DOMAIN, "INFO", otherMachineLines); + push(OTHER_TENANT_DOMAIN, "INFO", otherTenantLines); + + awaitLines(MACHINE_ID, DEVICE_LINES + TIED_LINES); + } + + @Test + void returnsEveryLineOfTheDeviceAndNothingFromOtherDevicesOrTenants() { + List lines = walk(MACHINE_ID, window(), 100); + + assertThat(lines).hasSize(DEVICE_LINES + TIED_LINES); + assertThat(lines).extracting(DeviceLogEntry::getMessage).noneMatch(message -> message.startsWith("other-")); + assertThat(lines).extracting(DeviceLogEntry::getHostname).containsOnly(MACHINE_ID + ".local"); + } + + @Test + void pagesStayNewestFirstWithoutGapsOrDuplicatesWhenLinesShareATimestamp() { + // Two-line pages put a page boundary among the five lines that share one nanosecond + DeviceLogFilterCriteria aroundTie = DeviceLogFilterCriteria.builder() + .from(instant(TIED_NANOS - 5 * STEP_NANOS)) + .to(instant(TIED_NANOS + 5 * STEP_NANOS)) + .build(); + + List paged = walk(MACHINE_ID, aroundTie, 2); + List single = service.queryDeviceLogs(MACHINE_ID, aroundTie, page(500, null)).getItems(); + + assertThat(paged).hasSize(10 + TIED_LINES); + assertThat(paged).extracting(DeviceLogServiceIT::key).doesNotHaveDuplicates() + .containsExactlyElementsOf(single.stream().map(DeviceLogServiceIT::key).toList()); + assertThat(paged).extracting(DeviceLogEntry::getMessage).contains("tied-0", "tied-1", "tied-2", "tied-3", "tied-4"); + assertThat(paged).extracting(DeviceLogEntry::getTimestamp).isSortedAccordingTo(Comparator.reverseOrder()); + } + + @Test + void servesUpToFiveHundredLinesPerPageAndOneHundredByDefault() { + GenericQueryResult capped = service.queryDeviceLogs(MACHINE_ID, window(), page(1000, null)); + GenericQueryResult defaulted = service.queryDeviceLogs(MACHINE_ID, window(), page(null, null)); + + assertThat(capped.getItems()).hasSize(500); + assertThat(capped.getPageInfo().isHasNextPage()).isTrue(); + assertThat(defaulted.getItems()).hasSize(100); + } + + @Test + void filtersByLevelSearchAndTimeWindow() { + List errors = walk(MACHINE_ID, + DeviceLogFilterCriteria.builder().from(FROM).levels(List.of(DeviceLogLevel.ERROR)).build(), 500); + List failures = walk(MACHINE_ID, + DeviceLogFilterCriteria.builder().from(FROM).search("connection failed").build(), 500); + List lastSecond = walk(MACHINE_ID, + DeviceLogFilterCriteria.builder().from(instant(BASE_NANOS + 590 * STEP_NANOS)).build(), 500); + + assertThat(errors).hasSize(ERROR_LINES + TIED_ERROR_LINES) + .extracting(DeviceLogEntry::getLevel).containsOnly("ERROR"); + assertThat(failures).hasSize(ERROR_LINES) + .extracting(DeviceLogEntry::getMessage).allMatch(message -> message.contains("Connection FAILED")); + assertThat(lastSecond).hasSize(10) + .extracting(DeviceLogEntry::getMessage).first().isEqualTo("line-599 heartbeat ok"); + } + + @Test + void matchesSearchTextLiterallySoItCannotWidenTheQuery() { + List hostile = walk(MACHINE_ID, + DeviceLogFilterCriteria.builder().from(FROM).search("\"} or {job=~\".+").build(), 500); + List regexLooking = walk(MACHINE_ID, + DeviceLogFilterCriteria.builder().from(FROM).search("line-0.0").build(), 500); + List literal = walk(MACHINE_ID, + DeviceLogFilterCriteria.builder().from(FROM).search("line-010").build(), 500); + + assertThat(hostile).isEmpty(); + // Unescaped, "." would match line-000 through line-090 + assertThat(regexLooking).isEmpty(); + assertThat(literal).extracting(DeviceLogEntry::getMessage).containsExactly("line-010 heartbeat ok"); + } + + @Test + void pollingFromTheNewestLineReturnsItAgainWithEveryNewerLine() throws Exception { + long seenNanos = BASE_NANOS + 10 * STEP_NANOS; + push(TENANT_DOMAIN, "INFO", List.of(entry(seenNanos, "poll-seen", POLL_MACHINE_ID))); + awaitLines(POLL_MACHINE_ID, 1); + DeviceLogEntry newest = service.queryDeviceLogs(POLL_MACHINE_ID, window(), page(100, null)).getItems().get(0); + + push(TENANT_DOMAIN, "INFO", List.of( + entry(seenNanos + STEP_NANOS, "poll-new-1", POLL_MACHINE_ID), + entry(seenNanos + 2 * STEP_NANOS, "poll-new-2", POLL_MACHINE_ID))); + awaitLines(POLL_MACHINE_ID, 3); + + List polled = service.queryDeviceLogs(POLL_MACHINE_ID, + DeviceLogFilterCriteria.builder().from(newest.getTimestamp()).build(), page(100, null)).getItems(); + + assertThat(polled).extracting(DeviceLogEntry::getMessage).containsExactly("poll-new-2", "poll-new-1", "poll-seen"); + } + + private static List walk(String machineId, DeviceLogFilterCriteria filter, int pageSize) { + List lines = new ArrayList<>(); + String rawCursor = null; + for (int pages = 0; pages < 1000; pages++) { + GenericQueryResult result = service.queryDeviceLogs(machineId, filter, page(pageSize, rawCursor)); + lines.addAll(result.getItems()); + if (!result.getPageInfo().isHasNextPage()) { + return lines; + } + rawCursor = CursorCodec.decode(result.getPageInfo().getEndCursor()); + } + throw new AssertionError("Paging did not finish within 1000 pages"); + } + + private static DeviceLogFilterCriteria window() { + return DeviceLogFilterCriteria.builder().from(FROM).build(); + } + + private static CursorPaginationCriteria page(Integer limit, String rawCursor) { + return CursorPaginationCriteria.builder().limit(limit).cursor(rawCursor).build(); + } + + private static String key(DeviceLogEntry entry) { + return toNanos(entry.getTimestamp()) + "|" + entry.getMessage(); + } + + private static DeviceLogService newService() { + DeviceService deviceService = mock(DeviceService.class); + when(deviceService.findByMachineId(anyString())).thenReturn(Optional.of(mock(Machine.class))); + TenantIdProvider tenantIdProvider = mock(TenantIdProvider.class); + when(tenantIdProvider.getTenantId()).thenReturn(TENANT_ID); + TenantRepository tenantRepository = mock(TenantRepository.class); + when(tenantRepository.findById(TENANT_ID)) + .thenReturn(Optional.of(Tenant.builder().id(TENANT_ID).domain(TENANT_DOMAIN).build())); + return new DeviceLogService(lokiClient, deviceService, tenantIdProvider, tenantRepository); + } + + private static void awaitLines(String machineId, int expected) throws InterruptedException { + String query = DeviceLogService.buildQuery(TENANT_DOMAIN, machineId, new DeviceLogFilterCriteria()); + for (int attempt = 0; attempt < 50; attempt++) { + int found = lokiClient.queryRange(query, toNanos(FROM), toNanos(Instant.now()) + 1, expected + 1, + LokiDirection.BACKWARD).size(); + if (found >= expected) { + return; + } + Thread.sleep(200); + } + throw new AssertionError("Loki did not return " + expected + " lines for " + machineId); + } + + private static List entry(long timestampNanos, String message, String machineId) { + Map metadata = Map.of( + "machine_id", machineId, + "hostname", machineId + ".local", + "agent_ts", instant(timestampNanos).truncatedTo(ChronoUnit.MILLIS).toString(), + "count", "1"); + return List.of(String.valueOf(timestampNanos), message, metadata); + } + + private static void push(String tenantDomain, String level, List> values) throws Exception { + Map labels = Map.of("job", "agent-logs", "tenant_domain", tenantDomain, "level", level); + String body = JSON.writeValueAsString(Map.of("streams", List.of(Map.of("stream", labels, "values", values)))); + HttpRequest request = HttpRequest.newBuilder(URI.create(baseUrl() + "/loki/api/v1/push")) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(body)) + .build(); + HttpResponse response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString()); + assertThat(response.statusCode()).as(response.body()).isEqualTo(204); + } + + private static String baseUrl() { + return "http://" + LOKI.getHost() + ":" + LOKI.getMappedPort(3100); + } + + private static Instant instant(long nanos) { + return Instant.ofEpochSecond(0, nanos); + } + + private static long toNanos(Instant instant) { + return instant.getEpochSecond() * 1_000_000_000L + instant.getNano(); + } +} diff --git a/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java b/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java index 95878157b7..25de68ab1e 100644 --- a/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java +++ b/openframe-api-lib/src/test/java/com/openframe/api/service/device/DeviceLogServiceTest.java @@ -84,7 +84,7 @@ void pinsSelectorToTheTenantDomainAndEscapesFilters() { verify(lokiClient).queryRange( "{job=\"agent-logs\", tenant_domain=\"acme.openframe.ai\", level=~\"ERROR|WARN\"}" + " |~ \"(?i)a\\\"b\" | machine_id=\"machine-1\"", - FROM_NANOS, TO_NANOS + 1, 21, LokiDirection.BACKWARD); + FROM_NANOS, TO_NANOS + 1, 101, LokiDirection.BACKWARD); } @Test @@ -93,7 +93,16 @@ void defaultsToTheLastSevenDaysWithoutLevelOrSearchFilters() { verify(lokiClient).queryRange( "{job=\"agent-logs\", tenant_domain=\"acme.openframe.ai\"} | machine_id=\"machine-1\"", - TO_NANOS - Duration.ofDays(7).toNanos(), TO_NANOS + 1, 21, LokiDirection.BACKWARD); + TO_NANOS - Duration.ofDays(7).toNanos(), TO_NANOS + 1, 101, LokiDirection.BACKWARD); + } + + @Test + void clampsThePageSizeBetweenOneAndFiveHundredLines() { + service.queryDeviceLogs(MACHINE_ID, window(), page(1000, null)); + service.queryDeviceLogs(MACHINE_ID, window(), page(0, null)); + + verify(lokiClient).queryRange(anyString(), anyLong(), anyLong(), eq(501), eq(LokiDirection.BACKWARD)); + verify(lokiClient).queryRange(anyString(), anyLong(), anyLong(), eq(2), eq(LokiDirection.BACKWARD)); } @Test @@ -147,30 +156,55 @@ void firstPageReportsNextPageAndCursors() { assertThat(result.getItems()).extracting(DeviceLogEntry::getMessage).containsExactly("c", "b"); assertThat(result.getPageInfo().isHasNextPage()).isTrue(); assertThat(result.getPageInfo().isHasPreviousPage()).isFalse(); - assertThat(result.getPageInfo().getStartCursor()).isEqualTo(CursorCodec.encode("300:1")); - assertThat(result.getPageInfo().getEndCursor()).isEqualTo(CursorCodec.encode("200:1")); + assertThat(result.getPageInfo().getStartCursor()).isEqualTo(CursorCodec.encode("300")); + assertThat(result.getPageInfo().getEndCursor()).isEqualTo(CursorCodec.encode("200")); } @Test - void nextPageSkipsEntriesAlreadyReturnedAtTheCursorTimestamp() { + void nextPageEndsJustBeforeTheCursorTimestamp() { long cursorNanos = TO_NANOS - 500; - when(lokiClient.queryRange(anyString(), eq(FROM_NANOS), eq(cursorNanos + 1), eq(4), eq(LokiDirection.BACKWARD))) - .thenReturn(List.of(entry(cursorNanos, "b1"), entry(cursorNanos, "b2"), entry(cursorNanos - 1, "a"))); + when(lokiClient.queryRange(anyString(), eq(FROM_NANOS), eq(cursorNanos), eq(3), eq(LokiDirection.BACKWARD))) + .thenReturn(List.of(entry(cursorNanos - 1, "b"), entry(cursorNanos - 2, "a"))); GenericQueryResult result = - service.queryDeviceLogs(MACHINE_ID, window(), page(2, cursorNanos + ":1")); + service.queryDeviceLogs(MACHINE_ID, window(), page(2, String.valueOf(cursorNanos))); - assertThat(result.getItems()).extracting(DeviceLogEntry::getMessage).containsExactly("b2", "a"); - assertThat(result.getItems()).extracting(DeviceLogEntry::getCursor) - .containsExactly(CursorCodec.encode(cursorNanos + ":2"), CursorCodec.encode((cursorNanos - 1) + ":1")); + assertThat(result.getItems()).extracting(DeviceLogEntry::getMessage).containsExactly("b", "a"); assertThat(result.getPageInfo().isHasNextPage()).isFalse(); assertThat(result.getPageInfo().isHasPreviousPage()).isTrue(); } + @Test + void endsAPageBeforeLinesThatShareTheCutTimestamp() { + // Loki cut the lines at 200 ns at the limit, keeping whichever it chose + when(lokiClient.queryRange(anyString(), anyLong(), anyLong(), eq(3), eq(LokiDirection.BACKWARD))) + .thenReturn(List.of(entry(300, "c"), entry(200, "b1"), entry(200, "b2"))); + + GenericQueryResult result = service.queryDeviceLogs(MACHINE_ID, window(), page(2, null)); + + assertThat(result.getItems()).extracting(DeviceLogEntry::getMessage).containsExactly("c"); + assertThat(result.getPageInfo().isHasNextPage()).isTrue(); + assertThat(result.getPageInfo().getEndCursor()).isEqualTo(CursorCodec.encode("300")); + } + + @Test + void returnsEveryLineOfATimestampThatFillsTheWholePage() { + when(lokiClient.queryRange(anyString(), anyLong(), anyLong(), eq(3), eq(LokiDirection.BACKWARD))) + .thenReturn(List.of(entry(200, "a"), entry(200, "b"), entry(200, "c"))); + when(lokiClient.queryRange(anyString(), eq(200L), eq(201L), eq(5000), eq(LokiDirection.BACKWARD))) + .thenReturn(List.of(entry(200, "a"), entry(200, "b"), entry(200, "c"), entry(200, "d"))); + + GenericQueryResult result = service.queryDeviceLogs(MACHINE_ID, window(), page(2, null)); + + assertThat(result.getItems()).extracting(DeviceLogEntry::getMessage).containsExactly("a", "b", "c", "d"); + assertThat(result.getPageInfo().isHasNextPage()).isTrue(); + assertThat(result.getPageInfo().getEndCursor()).isEqualTo(CursorCodec.encode("200")); + } + @Test void cursorBeforeTheWindowReturnsAnEmptyPageWithoutQuerying() { GenericQueryResult result = - service.queryDeviceLogs(MACHINE_ID, window(), page(2, (FROM_NANOS - 1) + ":1")); + service.queryDeviceLogs(MACHINE_ID, window(), page(2, String.valueOf(FROM_NANOS - 1))); assertThat(result.getItems()).isEmpty(); assertThat(result.getPageInfo().isHasNextPage()).isFalse(); diff --git a/openframe-api-service-core/src/main/resources/schema/device-log.graphqls b/openframe-api-service-core/src/main/resources/schema/device-log.graphqls index 8fd6160139..966476728e 100644 --- a/openframe-api-service-core/src/main/resources/schema/device-log.graphqls +++ b/openframe-api-service-core/src/main/resources/schema/device-log.graphqls @@ -4,7 +4,7 @@ extend type Query { deviceLogs( machineId: String! filter: DeviceLogFilterInput - first: Int + first: Int # Lines per page: 1-500, default 100 after: String ): DeviceLogConnection! } diff --git a/openframe-data-loki/pom.xml b/openframe-data-loki/pom.xml index 0c59634269..59a7ef3575 100644 --- a/openframe-data-loki/pom.xml +++ b/openframe-data-loki/pom.xml @@ -35,5 +35,10 @@ spring-boot-starter-test test + + org.testcontainers + testcontainers + test + diff --git a/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LokiClientIT.java b/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LokiClientIT.java new file mode 100644 index 0000000000..0f86dc9089 --- /dev/null +++ b/openframe-data-loki/src/test/java/com/openframe/data/loki/client/LokiClientIT.java @@ -0,0 +1,134 @@ +package com.openframe.data.loki.client; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.openframe.data.loki.model.LokiDirection; +import com.openframe.data.loki.model.LokiLogEntry; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.springframework.web.client.RestClient; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.utility.DockerImageName; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * {@link LokiClient} against a real Loki, the version the shared clusters run. + */ +class LokiClientIT { + + private static final GenericContainer LOKI = new GenericContainer<>(DockerImageName.parse("grafana/loki:3.7.3")) + .withExposedPorts(3100) + .waitingFor(Wait.forHttp("/ready").forPort(3100).forStatusCode(200) + .withStartupTimeout(Duration.ofMinutes(2))); + + private static final ObjectMapper JSON = new ObjectMapper(); + private static final long BASE_NANOS = toNanos(Instant.now().minus(Duration.ofMinutes(30)).truncatedTo(ChronoUnit.MILLIS)); + private static final long ONE_SECOND = 1_000_000_000L; + + private static LokiClient client; + + @BeforeAll + static void startLoki() { + LOKI.start(); + client = new LokiClient(RestClient.builder().baseUrl(baseUrl()).build()); + } + + @Test + void sendsQueriesWithPlusSignsQuotesAndBracesIntact() throws Exception { + String line = "retry a+b \"quoted\" {braces} | pipe"; + push(Map.of("job", "encoding"), List.of(entry(BASE_NANOS, line, Map.of()))); + + List entries = awaitEntries("{job=\"encoding\"} |= " + LogQl.quote(line), + BASE_NANOS, BASE_NANOS + 1, LokiDirection.BACKWARD, 1); + + assertThat(entries).extracting(LokiLogEntry::line).containsExactly(line); + } + + @Test + void mergesStreamsInTimestampOrderAndCarriesStructuredMetadata() throws Exception { + long start = BASE_NANOS + ONE_SECOND; + push(Map.of("job", "merge", "level", "INFO"), List.of( + entry(start, "first", Map.of("machine_id", "m-1")), + entry(start + 2, "third", Map.of("machine_id", "m-1")))); + push(Map.of("job", "merge", "level", "ERROR"), List.of( + entry(start + 1, "second", Map.of("machine_id", "m-1")))); + + List backward = awaitEntries("{job=\"merge\"}", start, start + 3, LokiDirection.BACKWARD, 3); + List forward = client.queryRange("{job=\"merge\"}", start, start + 3, 10, LokiDirection.FORWARD); + + assertThat(backward).extracting(LokiLogEntry::line).containsExactly("third", "second", "first"); + assertThat(forward).extracting(LokiLogEntry::line).containsExactly("first", "second", "third"); + assertThat(backward.get(1).labels()).containsEntry("level", "ERROR").containsEntry("machine_id", "m-1"); + assertThat(backward.get(1).timestampNanos()).isEqualTo(start + 1); + } + + @Test + void includesTheStartTimestampAndExcludesTheEndTimestamp() throws Exception { + // DeviceLogService's cursor depends on this: it ends the next page's query 1 ns after the cursor + long start = BASE_NANOS + 2 * ONE_SECOND; + push(Map.of("job", "bounds"), List.of( + entry(start, "at-start", Map.of()), + entry(start + 1, "middle", Map.of()), + entry(start + 2, "at-end", Map.of()))); + awaitEntries("{job=\"bounds\"}", start, start + 3, LokiDirection.BACKWARD, 3); + + List entries = client.queryRange("{job=\"bounds\"}", start, start + 2, 10, LokiDirection.BACKWARD); + + assertThat(entries).extracting(LokiLogEntry::line).containsExactly("middle", "at-start"); + } + + @Test + void wrapsRejectedQueriesInLokiQueryException() { + assertThatThrownBy(() -> client.queryRange("{job=", BASE_NANOS, BASE_NANOS + 1, 1, LokiDirection.BACKWARD)) + .isInstanceOf(LokiQueryException.class) + .hasMessageContaining("HTTP 400"); + } + + private static List awaitEntries(String query, long startNanos, long endNanos, + LokiDirection direction, int expected) throws InterruptedException { + List entries = List.of(); + for (int attempt = 0; attempt < 50; attempt++) { + entries = client.queryRange(query, startNanos, endNanos, 100, direction); + if (entries.size() >= expected) { + break; + } + Thread.sleep(200); + } + return entries; + } + + private static List entry(long timestampNanos, String line, Map metadata) { + String timestamp = String.valueOf(timestampNanos); + return metadata.isEmpty() ? List.of(timestamp, line) : List.of(timestamp, line, metadata); + } + + private static void push(Map labels, List> values) throws Exception { + String body = JSON.writeValueAsString(Map.of("streams", List.of(Map.of("stream", labels, "values", values)))); + HttpRequest request = HttpRequest.newBuilder(URI.create(baseUrl() + "/loki/api/v1/push")) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(body)) + .build(); + HttpResponse response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString()); + assertThat(response.statusCode()).as(response.body()).isEqualTo(204); + } + + private static String baseUrl() { + return "http://" + LOKI.getHost() + ":" + LOKI.getMappedPort(3100); + } + + private static long toNanos(Instant instant) { + return instant.getEpochSecond() * ONE_SECOND + instant.getNano(); + } +}