Add log cache log streaming support

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheEndpoints.java

@@ -16,19 +16,39 @@
 
 package org.cloudfoundry.reactor.logcache.v1;
 
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
+import org.cloudfoundry.AbstractCloudFoundryException;
+import org.cloudfoundry.logcache.v1.Envelope;
+import org.cloudfoundry.logcache.v1.EnvelopeType;
 import org.cloudfoundry.logcache.v1.InfoRequest;
 import org.cloudfoundry.logcache.v1.InfoResponse;
 import org.cloudfoundry.logcache.v1.MetaRequest;
 import org.cloudfoundry.logcache.v1.MetaResponse;
 import org.cloudfoundry.logcache.v1.ReadRequest;
 import org.cloudfoundry.logcache.v1.ReadResponse;
+import org.cloudfoundry.logcache.v1.TailLogsRequest;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
 
 final class ReactorLogCacheEndpoints extends AbstractLogCacheOperations {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(ReactorLogCacheEndpoints.class);
+
+    private static final int MAX_RETRIES = 5;
+
+    private static final Duration MAX_BACKOFF = Duration.ofSeconds(10);
+
     ReactorLogCacheEndpoints(
             ConnectionContext connectionContext,
             Mono<String> root,
@@ -48,4 +68,116 @@ Mono<MetaResponse> meta(MetaRequest request) {
     Mono<ReadResponse> read(ReadRequest request) {
         return get(request, ReadResponse.class, "read", request.getSourceId()).checkpoint();
     }
+
+    /**
+     * Continuously polls Log Cache and emits new {@link Envelope}s as they arrive.
+     *
+     * <p>Mirrors the Go {@code logcache.Walk()} / {@code WalkWithDelay} semantics:
+     * <ol>
+     *   <li>On every subscription, start a cursor at {@code startTime} (defaults to now minus 5 s
+     *       in nanoseconds). The cursor is per subscription, so subscribers never interfere.</li>
+     *   <li>Issue {@code GET /api/v1/read/{sourceId}?start_time=cursor}.</li>
+     *   <li>Log Cache is eventually consistent, so envelopes younger than {@code delay} are
+     *       withheld and picked up by a later poll. The rest are emitted in ascending timestamp
+     *       order and the cursor advances to {@code lastTimestamp + 1}.</li>
+     *   <li>When nothing can be emitted, wait {@code pollInterval} before polling again.</li>
+     *   <li>Transient failures (I/O errors, HTTP 429 and 5xx) are retried with exponential
+     *       backoff, up to {@value #MAX_RETRIES} times in a row; other errors terminate the
+     *       stream.</li>
+     * </ol>
+     * The stream never completes on its own – the caller cancels the subscription to stop.
+     */
+    Flux<Envelope> logsTail(TailLogsRequest request) {
+        List<EnvelopeType> envelopeTypes =
+                request.getEnvelopeTypes() != null
+                        ? request.getEnvelopeTypes()
+                        : Collections.emptyList();
+
+        return Mono.fromSupplier(
+                        () ->
+                                request.getStartTime() != null
+                                        ? request.getStartTime()
+                                        : toNanos(Instant.now().minusSeconds(5)))
+                .flatMap(cursor -> poll(request, envelopeTypes, cursor))
+                .expand(
+                        page ->
+                                page.envelopes.isEmpty()
+                                        ? Mono.delay(request.getPollInterval())
+                                                .then(poll(request, envelopeTypes, page.cursor))
+                                        : poll(request, envelopeTypes, page.nextCursor()))
+                .concatMapIterable(page -> page.envelopes);
+    }
+
+    private Mono<Page> poll(
+            TailLogsRequest request, List<EnvelopeType> envelopeTypes, long cursor) {
+        ReadRequest.Builder builder =
+                ReadRequest.builder().sourceId(request.getSourceId()).startTime(cursor);
+        if (!envelopeTypes.isEmpty()) {
+            builder.envelopeTypes(envelopeTypes);
+        }
+        if (request.getNameFilter() != null && !request.getNameFilter().isEmpty()) {
+            builder.nameFilter(request.getNameFilter());
+        }
+
+        return read(builder.build())
+                .retryWhen(
+                        Retry.backoff(MAX_RETRIES, request.getPollInterval())
+                                .maxBackoff(MAX_BACKOFF)
+                                .filter(ReactorLogCacheEndpoints::isTransient)
+                                .doBeforeRetry(
+                                        signal ->
+                                                LOGGER.warn(
+                                                        "Transient error polling Log Cache, retry"
+                                                                + " {}/{}",
+                                                        signal.totalRetries() + 1,
+                                                        MAX_RETRIES,
+                                                        signal.failure()))
+                                .onRetryExhaustedThrow((spec, signal) -> signal.failure()))
+                .map(response -> toPage(response, cursor, request.getDelay()));
+    }
+
+    private static Page toPage(ReadResponse response, long cursor, Duration delay) {
+        List<Envelope> raw =
+                response.getEnvelopes() != null
+                        ? response.getEnvelopes().getBatch()
+                        : Collections.emptyList();
+        long newestAllowed = toNanos(Instant.now().minus(delay));
+
+        List<Envelope> ready = new ArrayList<>();
+        for (Envelope envelope : raw) {
+            if (envelope.getTimestamp() != null && envelope.getTimestamp() <= newestAllowed) {
+                ready.add(envelope);
+            }
+        }
+        ready.sort(Comparator.comparingLong(Envelope::getTimestamp));
+        return new Page(cursor, ready);
+    }
+
+    private static boolean isTransient(Throwable t) {
+        if (t instanceof AbstractCloudFoundryException) {
+            int status = ((AbstractCloudFoundryException) t).getStatusCode();
+            return status == 429 || status >= 500;
+        }
+        return true;
+    }
+
+    private static long toNanos(Instant instant) {
+        return instant.getEpochSecond() * 1_000_000_000L + instant.getNano();
+    }
+
+    private static final class Page {
+
+        private final long cursor;
+
+        private final List<Envelope> envelopes;
+
+        private Page(long cursor, List<Envelope> envelopes) {
+            this.cursor = cursor;
+            this.envelopes = envelopes;
+        }
+
+        private long nextCursor() {
+            return this.envelopes.get(this.envelopes.size() - 1).getTimestamp() + 1;
+        }
+    }
 }

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/_ReactorLogCacheClient.java

@@ -16,16 +16,19 @@
 
 package org.cloudfoundry.reactor.logcache.v1;
 
+import org.cloudfoundry.logcache.v1.Envelope;
 import org.cloudfoundry.logcache.v1.InfoRequest;
 import org.cloudfoundry.logcache.v1.InfoResponse;
 import org.cloudfoundry.logcache.v1.LogCacheClient;
 import org.cloudfoundry.logcache.v1.MetaRequest;
 import org.cloudfoundry.logcache.v1.MetaResponse;
 import org.cloudfoundry.logcache.v1.ReadRequest;
 import org.cloudfoundry.logcache.v1.ReadResponse;
+import org.cloudfoundry.logcache.v1.TailLogsRequest;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
 import org.immutables.value.Value;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.net.URI;
@@ -53,6 +56,11 @@ public Mono<ReadResponse> read(ReadRequest request) {
         return getReactorLogCacheEndpoints().read(request);
     }
 
+    @Override
+    public Flux<Envelope> logsTail(TailLogsRequest request) {
+        return getReactorLogCacheEndpoints().logsTail(request);
+    }
+
     /**
      * The connection context
      */

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/util/ErrorPayloadMappers.java

@@ -74,8 +74,11 @@ public static ErrorPayloadMapper fallback() {
                                 ByteBufFlux body =
                                         ByteBufFlux.fromInbound(connection.inbound().receive());
 
+                                // An error response without a body must still surface as an error
+                                // instead of silently completing empty.
                                 return body.aggregate()
                                         .asString()
+                                        .defaultIfEmpty("")
                                         .doFinally(signalType -> connection.channel().close())
                                         .flatMap(
                                                 payload -> {

cloudfoundry-client-reactor/src/test/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheClientTest.java

@@ -17,7 +17,9 @@
 package org.cloudfoundry.reactor.logcache.v1;
 
 import static io.netty.handler.codec.http.HttpMethod.GET;
+import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
 import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.SERVICE_UNAVAILABLE;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
@@ -27,6 +29,7 @@
 import java.net.URI;
 import java.time.Duration;
 import java.util.Collections;
+import org.cloudfoundry.UnknownCloudFoundryException;
 import org.cloudfoundry.logcache.v1.Envelope;
 import org.cloudfoundry.logcache.v1.EnvelopeBatch;
 import org.cloudfoundry.logcache.v1.Gauge;
@@ -40,13 +43,15 @@
 import org.cloudfoundry.logcache.v1.Metric;
 import org.cloudfoundry.logcache.v1.ReadRequest;
 import org.cloudfoundry.logcache.v1.ReadResponse;
+import org.cloudfoundry.logcache.v1.TailLogsRequest;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.DefaultConnectionContext;
 import org.cloudfoundry.reactor.InteractionContext;
 import org.cloudfoundry.reactor.RootProvider;
 import org.cloudfoundry.reactor.TestRequest;
 import org.cloudfoundry.reactor.TestResponse;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
@@ -350,4 +355,94 @@ void read() {
                 .expectComplete()
                 .verify(Duration.ofSeconds(5));
     }
+
+    @Test
+    void logsTailOrdersEnvelopesAndAdvancesCursor() {
+        mockTailRead(10, "fixtures/logcache.v1/GET_read_tail_page1_response.json");
+        mockTailRead(21, "fixtures/logcache.v1/GET_read_tail_page2_response.json");
+
+        this.logCacheEndpoints
+                .logsTail(tailRequest())
+                .take(3)
+                .map(envelope -> envelope.getTimestamp())
+                .as(StepVerifier::create)
+                .expectNext(10L, 20L, 30L)
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void logsTailRetriesTransientErrors() {
+        mockRequest(
+                InteractionContext.builder()
+                        .request(tailGet(10))
+                        .response(TestResponse.builder().status(SERVICE_UNAVAILABLE).build())
+                        .build());
+        mockTailRead(10, "fixtures/logcache.v1/GET_read_tail_page2_response.json");
+
+        this.logCacheEndpoints
+                .logsTail(tailRequest())
+                .take(1)
+                .map(envelope -> envelope.getTimestamp())
+                .as(StepVerifier::create)
+                .expectNext(30L)
+                .expectComplete()
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void logsTailFailsOnClientError() {
+        mockRequest(
+                InteractionContext.builder()
+                        .request(tailGet(10))
+                        .response(TestResponse.builder().status(NOT_FOUND).build())
+                        .build());
+
+        this.logCacheEndpoints
+                .logsTail(tailRequest())
+                .as(StepVerifier::create)
+                .expectError(UnknownCloudFoundryException.class)
+                .verify(Duration.ofSeconds(5));
+    }
+
+    @Test
+    void logsTailUsesIndependentCursorPerSubscription() {
+        mockTailRead(10, "fixtures/logcache.v1/GET_read_tail_page1_response.json");
+        mockTailRead(10, "fixtures/logcache.v1/GET_read_tail_page1_response.json");
+
+        Flux<Long> tail =
+                this.logCacheEndpoints.logsTail(tailRequest()).take(1).map(e -> e.getTimestamp());
+
+        StepVerifier.create(tail).expectNext(10L).expectComplete().verify(Duration.ofSeconds(5));
+        StepVerifier.create(tail).expectNext(10L).expectComplete().verify(Duration.ofSeconds(5));
+    }
+
+    private static TailLogsRequest tailRequest() {
+        return TailLogsRequest.builder()
+                .sourceId("test-source-id")
+                .startTime(10L)
+                .pollInterval(Duration.ofMillis(10))
+                .delay(Duration.ZERO)
+                .build();
+    }
+
+    private static TestRequest tailGet(long startTime) {
+        return TestRequest.builder()
+                .method(GET)
+                .path("/api/v1/read/test-source-id?start_time=" + startTime)
+                .build();
+    }
+
+    private void mockTailRead(long startTime, String payload) {
+        mockRequest(
+                InteractionContext.builder()
+                        .request(tailGet(startTime))
+                        .response(
+                                TestResponse.builder()
+                                        .status(OK)
+                                        .contentType("application/json")
+                                        .payload(payload)
+                                        .build())
+                        .build());
+    }
 }

cloudfoundry-client-reactor/src/test/resources/fixtures/logcache.v1/GET_read_tail_page1_response.json

@@ -0,0 +1,22 @@
+{
+  "envelopes" : {
+    "batch" : [
+      {
+        "log" : {
+          "payload" : "bXNnMg==",
+          "type" : "OUT"
+        },
+        "source_id" : "test-source-id",
+        "timestamp" : "20"
+      },
+      {
+        "log" : {
+          "payload" : "bXNnMQ==",
+          "type" : "ERR"
+        },
+        "source_id" : "test-source-id",
+        "timestamp" : "10"
+      }
+    ]
+  }
+}

cloudfoundry-client-reactor/src/test/resources/fixtures/logcache.v1/GET_read_tail_page2_response.json

@@ -0,0 +1,14 @@
+{
+  "envelopes" : {
+    "batch" : [
+      {
+        "log" : {
+          "payload" : "bXNnMw==",
+          "type" : "OUT"
+        },
+        "source_id" : "test-source-id",
+        "timestamp" : "30"
+      }
+    ]
+  }
+}

cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/LogCacheClient.java

@@ -16,6 +16,7 @@
 
 package org.cloudfoundry.logcache.v1;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /**
@@ -46,4 +47,17 @@ public interface LogCacheClient {
      * @return the read response
      */
     Mono<ReadResponse> read(ReadRequest request);
+
+    /**
+     * Continuously polls the Log Cache /api/v1/read endpoint and streams new {@link Envelope}s
+     * as they appear. This is the Java equivalent of the Go {@code logcache.Walk()} API and
+     * {@code cf tail --follow}.
+     * <p>
+     * The returned {@link Flux} will never complete on its own – unsubscribe (or cancel) it to
+     * stop streaming.
+     *
+     * @param request the tail request (source id, optional filters, poll interval)
+     * @return an infinite stream of envelopes
+     */
+    Flux<Envelope> logsTail(TailLogsRequest request);
 }