diff --git a/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheEndpoints.java b/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheEndpoints.java index 2e68c52538b..71dc04e8686 100644 --- a/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheEndpoints.java +++ b/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 root, @@ -48,4 +68,116 @@ Mono meta(MetaRequest request) { Mono 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. + * + *

Mirrors the Go {@code logcache.Walk()} / {@code WalkWithDelay} semantics: + *

    + *
  1. 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.
    2. + *
  2. Issue {@code GET /api/v1/read/{sourceId}?start_time=cursor}.
    3. + *
  3. 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}.
    4. + *
  4. When nothing can be emitted, wait {@code pollInterval} before polling again.
    5. + *
  5. 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.
    6. + *
+ * The stream never completes on its own – the caller cancels the subscription to stop. + */ + Flux logsTail(TailLogsRequest request) { + List 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 poll( + TailLogsRequest request, List 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 raw = + response.getEnvelopes() != null + ? response.getEnvelopes().getBatch() + : Collections.emptyList(); + long newestAllowed = toNanos(Instant.now().minus(delay)); + + List 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 envelopes; + + private Page(long cursor, List envelopes) { + this.cursor = cursor; + this.envelopes = envelopes; + } + + private long nextCursor() { + return this.envelopes.get(this.envelopes.size() - 1).getTimestamp() + 1; + } + } } diff --git a/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/_ReactorLogCacheClient.java b/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/_ReactorLogCacheClient.java index d9460476ea4..7f4938f1f79 100644 --- a/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/_ReactorLogCacheClient.java +++ b/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/_ReactorLogCacheClient.java @@ -16,6 +16,7 @@ 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; @@ -23,9 +24,11 @@ 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 read(ReadRequest request) { return getReactorLogCacheEndpoints().read(request); } + @Override + public Flux logsTail(TailLogsRequest request) { + return getReactorLogCacheEndpoints().logsTail(request); + } + /** * The connection context */ diff --git a/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/util/ErrorPayloadMappers.java b/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/util/ErrorPayloadMappers.java index 282b5ec2e61..4519ca5560d 100644 --- a/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/util/ErrorPayloadMappers.java +++ b/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 -> { diff --git a/cloudfoundry-client-reactor/src/test/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheClientTest.java b/cloudfoundry-client-reactor/src/test/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheClientTest.java index 740461dc469..abe5181d6fc 100644 --- a/cloudfoundry-client-reactor/src/test/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheClientTest.java +++ b/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,6 +43,7 @@ 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; @@ -47,6 +51,7 @@ 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 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()); + } } diff --git a/cloudfoundry-client-reactor/src/test/resources/fixtures/logcache.v1/GET_read_tail_page1_response.json b/cloudfoundry-client-reactor/src/test/resources/fixtures/logcache.v1/GET_read_tail_page1_response.json new file mode 100644 index 00000000000..cf925863e6a --- /dev/null +++ b/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" + } + ] + } +} diff --git a/cloudfoundry-client-reactor/src/test/resources/fixtures/logcache.v1/GET_read_tail_page2_response.json b/cloudfoundry-client-reactor/src/test/resources/fixtures/logcache.v1/GET_read_tail_page2_response.json new file mode 100644 index 00000000000..b4bca11a1ba --- /dev/null +++ b/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" + } + ] + } +} diff --git a/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/LogCacheClient.java b/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/LogCacheClient.java index e455db220a4..82466dfdf31 100644 --- a/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/LogCacheClient.java +++ b/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 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}. + *

+ * 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 logsTail(TailLogsRequest request); } diff --git a/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java b/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java new file mode 100644 index 00000000000..34dd8125405 --- /dev/null +++ b/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java @@ -0,0 +1,76 @@ +/* + * Copyright 2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.cloudfoundry.logcache.v1; + +import org.cloudfoundry.Nullable; +import org.immutables.value.Value; + +import java.time.Duration; +import java.util.List; + +/** + * The request options for the Log Cache tail (streaming follow) operation. + * This continuously polls the Log Cache /api/v1/read endpoint, emitting new envelopes + * as they appear – equivalent to {@code cf tail --follow} or the Go {@code logcache.Walk()} API. + */ +@Value.Immutable +abstract class _TailLogsRequest { + + /** + * The source id (application guid or service guid) to stream logs for. + */ + abstract String getSourceId(); + + /** + * Optional start time (UNIX nanoseconds). Defaults to "now – 5 seconds" when not set. + */ + @Nullable + abstract Long getStartTime(); + + /** + * Optional envelope type filter. + */ + @Nullable + abstract List getEnvelopeTypes(); + + /** + * Optional regex name filter (requires Log Cache ≥ 2.1.0). + */ + @Nullable + abstract String getNameFilter(); + + /** + * How long to wait between successive polls when no new envelopes are available. + * Defaults to 250 ms (matching the Go client's {@code AlwaysRetryBackoff}). + */ + @Value.Default + Duration getPollInterval() { + return Duration.ofMillis(250); + } + + /** + * Log Cache is eventually consistent: envelopes may become visible out of order. Envelopes + * younger than this delay are therefore withheld and re-read on a later poll, so that late + * arrivals are not skipped (equivalent to the Go client's {@code WalkWithDelay}). + * Defaults to 2 seconds. + */ + @Value.Default + Duration getDelay() { + return Duration.ofSeconds(2); + } +} + diff --git a/cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/Applications.java b/cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/Applications.java index 56aba6af646..d15df7f2313 100644 --- a/cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/Applications.java +++ b/cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/Applications.java @@ -128,9 +128,9 @@ public interface Applications { /** * List the applications logs. - * Uses Log Cache under the hood when {@link ApplicationLogsRequest#getRecent()} is {@code true}. - * Log streaming still uses Doppler, which is not available in CF deployments following - * shared-nothing architecture. + * Uses Log Cache under the hood: recent logs are read when {@link ApplicationLogsRequest#getRecent()} + * is {@code true}, otherwise logs are continuously streamed (like {@code cf logs}). The returned + * {@link Flux} of a streaming request never completes on its own – cancel it to stop. * * @param request the application logs request * @return the applications logs diff --git a/cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/DefaultApplications.java b/cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/DefaultApplications.java index c56ee52554a..ec4e7a0d6d1 100644 --- a/cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/DefaultApplications.java +++ b/cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/DefaultApplications.java @@ -155,8 +155,10 @@ import org.cloudfoundry.doppler.RecentLogsRequest; import org.cloudfoundry.doppler.StreamRequest; import org.cloudfoundry.logcache.v1.EnvelopeBatch; +import org.cloudfoundry.logcache.v1.EnvelopeType; import org.cloudfoundry.logcache.v1.LogCacheClient; import org.cloudfoundry.logcache.v1.ReadRequest; +import org.cloudfoundry.logcache.v1.TailLogsRequest; import org.cloudfoundry.operations.util.OperationsLogging; import org.cloudfoundry.util.DateUtils; import org.cloudfoundry.util.DelayTimeoutException; @@ -559,22 +561,25 @@ public Flux logs(LogsRequest request) { @Override public Flux logs(ApplicationLogsRequest request) { + Mono applicationId = + Mono.zip(this.cloudFoundryClient, this.spaceId) + .flatMap( + function( + (cloudFoundryClient, spaceId) -> + getApplicationId( + cloudFoundryClient, + request.getName(), + spaceId))); if (request.getRecent() != null && request.getRecent()) { - return Mono.zip(this.cloudFoundryClient, this.spaceId) - .flatMap( - function( - (cloudFoundryClient, spaceId) -> - getApplicationId( - cloudFoundryClient, - request.getName(), - spaceId))) - .flatMapMany( - applicationId -> getLogsLogCache(this.logCacheClient, applicationId)) + return applicationId + .flatMapMany(id -> getLogsLogCache(this.logCacheClient, id)) .transform(OperationsLogging.log("Get Application Logs")) .checkpoint(); } else { - return logs(LogsRequest.builder().name(request.getName()).recent(false).build()) - .map(DefaultApplications::toApplicationLog); + return applicationId + .flatMapMany(id -> getLogsTailLogCache(this.logCacheClient, id)) + .transform(OperationsLogging.log("Get Application Logs")) + .checkpoint(); } } @@ -1656,6 +1661,39 @@ private static Flux getLogsLogCache( .build()); } + private static Flux getLogsTailLogCache( + Mono logCacheClient, String applicationId) { + return logCacheClient + .flatMapMany( + client -> + client.logsTail( + TailLogsRequest.builder() + .sourceId(applicationId) + .envelopeTypes( + Collections.singletonList(EnvelopeType.LOG)) + .build())) + .filter(e -> e.getLog() != null) + .map( + envelope -> + ApplicationLog.builder() + .sourceId( + Optional.ofNullable(envelope.getSourceId()) + .orElse("")) + .sourceType( + envelope.getTags().getOrDefault("source_type", "")) + .instanceId( + Optional.ofNullable(envelope.getInstanceId()) + .orElse("")) + .message(envelope.getLog().getPayloadAsText()) + .timestamp( + Optional.ofNullable(envelope.getTimestamp()) + .orElse(0L)) + .logType( + ApplicationLogType.from( + envelope.getLog().getType().name())) + .build()); + } + @SuppressWarnings("unchecked") private static Map getMetadataRequest(EventEntity entity) { Map> metadata = @@ -2964,17 +3002,6 @@ private static Mono stopApplicationIfNotStopped( : Mono.just(resource); } - private static ApplicationLog toApplicationLog(LogMessage logMessage) { - return ApplicationLog.builder() - .sourceId(logMessage.getApplicationId()) - .sourceType(logMessage.getSourceType()) - .instanceId(logMessage.getSourceInstance()) - .message(logMessage.getMessage()) - .timestamp(logMessage.getTimestamp()) - .logType(ApplicationLogType.from(logMessage.getMessageType().name())) - .build(); - } - private static ApplicationDetail toApplicationDetail( List buildpacks, SummaryApplicationResponse summaryApplicationResponse, diff --git a/cloudfoundry-operations/src/test/java/org/cloudfoundry/operations/applications/DefaultApplicationsTest.java b/cloudfoundry-operations/src/test/java/org/cloudfoundry/operations/applications/DefaultApplicationsTest.java index f137ac2d84b..3fbffef1181 100644 --- a/cloudfoundry-operations/src/test/java/org/cloudfoundry/operations/applications/DefaultApplicationsTest.java +++ b/cloudfoundry-operations/src/test/java/org/cloudfoundry/operations/applications/DefaultApplicationsTest.java @@ -147,11 +147,13 @@ import org.cloudfoundry.doppler.StreamRequest; import org.cloudfoundry.logcache.v1.Envelope; import org.cloudfoundry.logcache.v1.EnvelopeBatch; +import org.cloudfoundry.logcache.v1.EnvelopeType; import org.cloudfoundry.logcache.v1.Log; import org.cloudfoundry.logcache.v1.LogCacheClient; import org.cloudfoundry.logcache.v1.LogType; import org.cloudfoundry.logcache.v1.ReadRequest; import org.cloudfoundry.logcache.v1.ReadResponse; +import org.cloudfoundry.logcache.v1.TailLogsRequest; import org.cloudfoundry.operations.AbstractOperationsTest; import org.cloudfoundry.util.DateUtils; import org.cloudfoundry.util.FluentMap; @@ -1396,6 +1398,79 @@ void logsLogCache() { .verify(Duration.ofSeconds(5)); } + @Test + void logsStreamLogCache() { + requestApplications( + this.cloudFoundryClient, + "test-application-name", + TEST_SPACE_ID, + "test-metadata-id"); + requestLogsTailLogCache( + this.logCacheClient, + tailRequest("test-metadata-id"), + tailEnvelope("msg1", LogType.OUT, 1L), + tailEnvelope("msg2", LogType.ERR, 2L), + tailEnvelope("msg3", LogType.OUT, 3L)); + + this.applications + .logs(ApplicationLogsRequest.builder().name("test-application-name").build()) + .take(3) + .as(StepVerifier::create) + .expectNextMatches( + log -> + log.getMessage().equals("msg1") + && log.getLogType() == ApplicationLogType.OUT + && log.getSourceId().equals("test-metadata-id") + && log.getTimestamp() == 1L) + .expectNextMatches( + log -> + log.getMessage().equals("msg2") + && log.getLogType() == ApplicationLogType.ERR) + .expectNextMatches(log -> log.getMessage().equals("msg3")) + .expectComplete() + .verify(Duration.ofSeconds(5)); + } + + @Test + void logsStreamLogCacheError() { + requestApplications( + this.cloudFoundryClient, + "test-application-name", + TEST_SPACE_ID, + "test-metadata-id"); + when(this.logCacheClient.logsTail(tailRequest("test-metadata-id"))) + .thenReturn(Flux.error(new RuntimeException("log-cache unavailable"))); + + this.applications + .logs( + ApplicationLogsRequest.builder() + .name("test-application-name") + .recent(false) + .build()) + .as(StepVerifier::create) + .expectErrorMatches( + t -> + t instanceof RuntimeException + && "log-cache unavailable".equals(t.getMessage())) + .verify(Duration.ofSeconds(5)); + } + + @Test + void logsStreamLogCacheApplicationDoesNotExist() { + requestApplicationsEmpty(this.cloudFoundryClient, "test-application-name", TEST_SPACE_ID); + + this.applications + .logs(ApplicationLogsRequest.builder().name("test-application-name").build()) + .as(StepVerifier::create) + .consumeErrorWith( + t -> + assertThat(t) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Application test-application-name does not exist")) + .verify(Duration.ofSeconds(5)); + } + @SuppressWarnings("deprecation") @Test void logsRecentNotSetDoppler() { @@ -5403,6 +5478,34 @@ private static void requestLogsRecentLogCache(LogCacheClient logCacheClient, Str .build())); } + private static TailLogsRequest tailRequest(String sourceId) { + return TailLogsRequest.builder() + .sourceId(sourceId) + .envelopeTypes(Collections.singletonList(EnvelopeType.LOG)) + .build(); + } + + private static Envelope tailEnvelope(String payload, LogType type, long timestamp) { + return Envelope.builder() + .sourceId("test-metadata-id") + .timestamp(timestamp) + .log( + Log.builder() + .payload( + Base64.getEncoder() + .encodeToString( + payload.getBytes(StandardCharsets.UTF_8))) + .type(type) + .build()) + .build(); + } + + private static void requestLogsTailLogCache( + LogCacheClient logCacheClient, TailLogsRequest tailRequest, Envelope... envelopes) { + when(logCacheClient.logsTail(tailRequest)) + .thenReturn(Flux.concat(Flux.just(envelopes), Flux.never())); + } + private static void requestLogsStream(DopplerClient dopplerClient, String applicationId) { when(dopplerClient.stream(StreamRequest.builder().applicationId(applicationId).build())) .thenReturn( diff --git a/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java b/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java index ed8319bafc0..7238e9651e9 100644 --- a/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java +++ b/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java @@ -34,6 +34,14 @@ import org.cloudfoundry.client.CloudFoundryClient; import org.cloudfoundry.client.v3.applications.ApplicationFeatureResource; import org.cloudfoundry.client.v3.applications.ListApplicationFeaturesRequest; +import org.cloudfoundry.logcache.v1.Envelope; +import org.cloudfoundry.logcache.v1.EnvelopeBatch; +import org.cloudfoundry.logcache.v1.EnvelopeType; +import org.cloudfoundry.logcache.v1.Log; +import org.cloudfoundry.logcache.v1.LogCacheClient; +import org.cloudfoundry.logcache.v1.LogType; +import org.cloudfoundry.logcache.v1.ReadRequest; +import org.cloudfoundry.logcache.v1.ReadResponse; import org.cloudfoundry.operations.applications.ApplicationDetail; import org.cloudfoundry.operations.applications.ApplicationEnvironments; import org.cloudfoundry.operations.applications.ApplicationEvent; @@ -106,6 +114,7 @@ public final class ApplicationsTest extends AbstractIntegrationTest { @Autowired private String serviceName; + @Autowired private LogCacheClient logCacheClient; @Autowired private CloudFoundryClient cloudFoundryClient; // To create a service in #pushBindService, the Service Broker must be installed first. @@ -538,6 +547,72 @@ public void logs() throws IOException { .verify(Duration.ofMinutes(5)); } + /** + * Exercise the LogCache client. Serves as a reference for using the logcache client, + * and will help with the transition to the new + * {@link org.cloudfoundry.operations.applications.Applications#logs(ApplicationLogsRequest)}. + */ + @Test + public void logCacheLogs() throws IOException { + String applicationName = this.nameFactory.getApplicationName(); + + createApplication( + this.cloudFoundryOperations, + new ClassPathResource("test-application.zip").getFile().toPath(), + applicationName, + false) + .then( + this.cloudFoundryOperations + .applications() + .get(GetApplicationRequest.builder().name(applicationName).build())) + .map(ApplicationDetail::getId) + .flatMapMany( + appGuid -> + this.logCacheClient.read( + ReadRequest.builder() + .sourceId(appGuid) + .envelopeType(EnvelopeType.LOG) + .limit(1) + .build())) + .map(ReadResponse::getEnvelopes) + .map(EnvelopeBatch::getBatch) + .flatMap(Flux::fromIterable) + .map(Envelope::getLog) + .map(Log::getType) + .next() + .as(StepVerifier::create) + .expectNext(LogType.OUT) + .expectComplete() + .verify(Duration.ofMinutes(5)); + } + + /** + * Streams application logs via {@code logs(ApplicationLogsRequest)} (Log Cache tailing). + */ + @Test + public void logsStream() throws IOException { + String applicationName = this.nameFactory.getApplicationName(); + + createApplication( + this.cloudFoundryOperations, + new ClassPathResource("test-application.zip").getFile().toPath(), + applicationName, + false) + .thenMany( + this.cloudFoundryOperations + .applications() + .logs( + ApplicationLogsRequest.builder() + .name(applicationName) + .build()) + .take(3)) + .map(ApplicationLog::getLogType) + .as(StepVerifier::create) + .expectNextCount(3) + .expectComplete() + .verify(Duration.ofMinutes(5)); + } + @Test public void pushBindServices() throws IOException { String applicationName = this.nameFactory.getApplicationName();