tenantDomains = new ConcurrentHashMap<>();
+
+ /**
+ * 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();
+ 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: 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;
+ if (after != null) {
+ if (after.timestampNanos() <= startNanos) {
+ return result(List.of(), false, true);
+ }
+ endNanos = Math.min(endNanos, after.timestampNanos());
+ }
+
+ String query = buildQuery(resolveTenantDomain(), machineId, criteria);
+ int pageSize = pageSize(page.getLimit());
+ log.debug("Querying device logs for machineId: {}, query: {}, start: {}, end: {}", machineId, query, startNanos, endNanos);
+
+ // 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(items, entries.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();
+ }
+
+ /**
+ * 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)
+ .orElse(null);
+ }
+
+ /**
+ * 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 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--;
+ }
+ 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) {
+ List items = new ArrayList<>(entries.size());
+ for (LokiLogEntry entry : entries) {
+ 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(entry.timestampNanos()).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();
+ }
+
+ /**
+ * 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");
+ }
+ }
+
+ 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/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