From 5bf80cf1b36b7df25e5512348ecad4a7b51f5242 Mon Sep 17 00:00:00 2001 From: Pascal Zimmermann Date: Tue, 3 Mar 2026 23:53:06 +0100 Subject: [PATCH 1/6] feat: Add Log Cache log streaming support --- .../logcache/v1/ReactorLogCacheEndpoints.java | 115 ++++++++++++++ .../logcache/v1/_ReactorLogCacheClient.java | 13 ++ .../logcache/v1/LogCacheClient.java | 22 +++ .../logcache/v1/_TailLogsRequest.java | 65 ++++++++ .../operations/applications/Applications.java | 28 +++- .../applications/DefaultApplications.java | 104 ++++++------ .../applications/DefaultApplicationsTest.java | 150 ++++++++++++++++++ 7 files changed, 436 insertions(+), 61 deletions(-) create mode 100644 cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java 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 2e68c52538..6df4093ed5 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,15 +16,23 @@ package org.cloudfoundry.reactor.logcache.v1; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; +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 reactor.core.publisher.Flux; import reactor.core.publisher.Mono; final class ReactorLogCacheEndpoints extends AbstractLogCacheOperations { @@ -48,4 +56,111 @@ Mono meta(MetaRequest request) { Mono read(ReadRequest request) { return get(request, ReadResponse.class, "read", request.getSourceId()).checkpoint(); } + + Mono recentLogs(ReadRequest request) { + return read(request); + } + + /** + * Continuously polls Log Cache and emits new {@link Envelope}s as they arrive. + * + *

Mirrors the Go {@code logcache.Walk()} / {@code cf tail --follow} semantics: + *

    + *
  1. Start the cursor at {@code startTime} (defaults to now − 5 s in + * nanoseconds).
    2. + *
  2. Issue {@code GET /api/v1/read/{sourceId}?start_time=cursor}.
    3. + *
  3. Emit every returned envelope in ascending timestamp order and advance + * the cursor to {@code lastTimestamp + 1}.
    4. + *
  4. When the batch is empty, wait {@code pollInterval} before the next poll.
    5. + *
  5. Repeat forever – the caller cancels the subscription to stop.
    6. + *
+ * Fully non-blocking: no {@code Thread.sleep}. + */ + Flux logsTail(TailLogsRequest request) { + long defaultStartNanos = (System.currentTimeMillis() - 5_000L) * 1_000_000L; + AtomicLong cursor = + new AtomicLong( + request.getStartTime() != null + ? request.getStartTime() + : defaultStartNanos); + + List envelopeTypes = + request.getEnvelopeTypes() != null + ? request.getEnvelopeTypes() + : Collections.emptyList(); + String nameFilter = request.getNameFilter(); + + /* + * Strategy (mirrors Go's logcache.Walk): + * – Mono.defer builds a fresh ReadRequest from the mutable cursor on every repetition. + * – The Mono returns either the sorted batch (non-empty) or an empty list. + * – flatMapMany turns each batch into a stream of individual Envelope items. + * – repeat() subscribes again after each completion. + * – When the batch was empty we insert a delay via Mono.delay before the next + * repetition so we do not hammer the server. We signal "empty" by returning + * a sentinel Mono (false = was empty, true = had data) and use + * repeatWhen to conditionally delay. + */ + return Flux.defer( + () -> { + // Build the read request from the current cursor position. + ReadRequest.Builder builder = + ReadRequest.builder() + .sourceId(request.getSourceId()) + .startTime(cursor.get()); + if (!envelopeTypes.isEmpty()) { + builder.envelopeTypes(envelopeTypes); + } + if (nameFilter != null && !nameFilter.isEmpty()) { + builder.nameFilter(nameFilter); + } + + return read(builder.build()) + .onErrorReturn(ReadResponse.builder().build()) + .flatMapMany( + resp -> { + List raw = + resp.getEnvelopes() != null + ? resp.getEnvelopes().getBatch() + : Collections.emptyList(); + + if (raw.isEmpty()) { + // Signal "no data" so repeatWhen can insert the + // back-off delay. + return Flux.empty(); + } + + // Sort ascending by timestamp and advance the + // cursor. + List sorted = new ArrayList<>(raw); + sorted.sort( + (a, b) -> + Long.compare( + a.getTimestamp() != null + ? a.getTimestamp() + : 0L, + b.getTimestamp() != null + ? b.getTimestamp() + : 0L)); + + Envelope last = sorted.get(sorted.size() - 1); + cursor.set( + (last.getTimestamp() != null + ? last.getTimestamp() + : cursor.get()) + + 1); + + return Flux.fromIterable(sorted); + }); + }) + // repeatWhen receives a Flux where each element is the count of items + // emitted in the previous cycle (0 = empty batch → insert delay). + .repeatWhen( + companion -> + companion.flatMap( + count -> + count == 0 + ? Mono.delay(request.getPollInterval()) + : Mono.just(count))); + } } 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 d9460476ea..0236da2188 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,16 @@ public Mono read(ReadRequest request) { return getReactorLogCacheEndpoints().read(request); } + @Override + public Mono recentLogs(ReadRequest request) { + return getReactorLogCacheEndpoints().recentLogs(request); + } + + @Override + public Flux logsTail(TailLogsRequest request) { + return getReactorLogCacheEndpoints().logsTail(request); + } + /** * The connection context */ 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 e455db220a..6b91461fe3 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,25 @@ public interface LogCacheClient { * @return the read response */ Mono read(ReadRequest request); + + /** + * Makes the Log Cache RecentLogs /api/v1/read request + * + * @param request the Recent Logs request + * @return the events from the recent logs + */ + Mono recentLogs(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 0000000000..1ca32d388a --- /dev/null +++ b/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java @@ -0,0 +1,65 @@ +/* + * Copyright 2013-2021 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); + } +} + 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 56aba6af64..5d95db5d01 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 @@ -17,6 +17,10 @@ package org.cloudfoundry.operations.applications; import org.cloudfoundry.doppler.LogMessage; +import org.cloudfoundry.logcache.v1.Envelope; +import org.cloudfoundry.logcache.v1.Log; +import org.cloudfoundry.logcache.v1.ReadRequest; +import org.cloudfoundry.logcache.v1.TailLogsRequest; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -126,11 +130,29 @@ public interface Applications { @Deprecated Flux logs(LogsRequest request); + /** + * List the applications logs from logCacheClient. + * If no messages are available, an empty Flux is returned. + * + * @param request the application logs request + * @return the applications logs + */ + Flux logsRecent(ReadRequest request); + + /** + * Continuously streams application log envelopes from Log Cache by repeatedly polling + * the {@code /api/v1/read} endpoint. The returned {@link Flux} is infinite – cancel it + * to stop streaming. This is the Java equivalent of {@code cf tail --follow}. + * + * @param request the tail request (source id, optional filters, poll interval) + * @return an infinite stream of envelopes + */ + Flux logsTail(TailLogsRequest request); + /** * 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. + * Only works with {@code Loggregator < 107.0}, shipped in {@code CFD < 24.3} + * and {@code TAS < 4.0}. * * @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 c56ee52554..23d1fb70cd 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.Log; 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; @@ -557,25 +559,39 @@ public Flux logs(LogsRequest request) { .checkpoint(); } + @Override + public Flux logsRecent(ReadRequest request) { + return getRecentLogsLogCache(this.logCacheClient, request) + .transform(OperationsLogging.log("Get Application Logs")) + .checkpoint(); + } + + @Override + public Flux logsTail(TailLogsRequest request) { + return this.logCacheClient + .flatMapMany(client -> client.logsTail(request)) + .transform(OperationsLogging.log("Tail Application Logs")) + .checkpoint(); + } + @Override public Flux logs(ApplicationLogsRequest request) { - 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)) - .transform(OperationsLogging.log("Get Application Logs")) - .checkpoint(); - } else { - return logs(LogsRequest.builder().name(request.getName()).recent(false).build()) - .map(DefaultApplications::toApplicationLog); - } + return logs(LogsRequest.builder() + .name(request.getName()) + .recent(request.getRecent()) + .build()) + .map( + logMessage -> + ApplicationLog.builder() + .sourceId(logMessage.getApplicationId()) + .sourceType(logMessage.getSourceType()) + .instanceId(logMessage.getSourceInstance()) + .message(logMessage.getMessage()) + .timestamp(logMessage.getTimestamp()) + .logType( + ApplicationLogType.from( + logMessage.getMessageType().name())) + .build()); } @Override @@ -1630,30 +1646,15 @@ private static Flux getLogs( } } - private static Flux getLogsLogCache( - Mono logCacheClient, String applicationId) { - return requestLogsRecentLogCache(logCacheClient, applicationId) + private static Flux getRecentLogsLogCache( + Mono logCacheClient, ReadRequest readRequest) { + return requestLogsRecentLogCache(logCacheClient, readRequest) + .map(EnvelopeBatch::getBatch) + .map(List::stream) + .flatMapIterable(envelopeStream -> envelopeStream.collect(Collectors.toList())) .filter(e -> e.getLog() != null) .sort(LOG_MESSAGE_COMPARATOR_LOG_CACHE) - .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()); + .map(org.cloudfoundry.logcache.v1.Envelope::getLog); } @SuppressWarnings("unchecked") @@ -2549,14 +2550,12 @@ private static Flux requestLogsRecent( RecentLogsRequest.builder().applicationId(applicationId).build())); } - private static Flux requestLogsRecentLogCache( - Mono logCacheClient, String applicationId) { - return logCacheClient - .flatMap( - client -> - client.read(ReadRequest.builder().sourceId(applicationId).build())) - .flatMap(response -> Mono.justOrEmpty(response.getEnvelopes())) - .flatMapIterable(EnvelopeBatch::getBatch); + private static Mono requestLogsRecentLogCache( + Mono logCacheClient, ReadRequest readRequest) { + return logCacheClient.flatMap( + client -> + client.recentLogs(readRequest) + .flatMap(response -> Mono.justOrEmpty(response.getEnvelopes()))); } private static Flux requestLogsStream( @@ -2964,17 +2963,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 f137ac2d84..af61a2363c 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 @@ -152,6 +152,7 @@ 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 +1397,71 @@ void logsLogCache() { .verify(Duration.ofSeconds(5)); } + @Test + void logsTailLogCache() { + TailLogsRequest tailRequest = TailLogsRequest.builder().sourceId("test-source-id").build(); + requestLogsTailLogCache(this.logCacheClient, tailRequest, "test-tail-payload"); + + this.applications + .logsTail(tailRequest) + .take(1) + .as(StepVerifier::create) + .expectNextMatches( + envelope -> + envelope.getLog() != null + && LogType.OUT.equals(envelope.getLog().getType())) + .expectComplete() + .verify(Duration.ofSeconds(5)); + } + + @Test + void logsTailLogCacheMultipleEnvelopes() { + TailLogsRequest tailRequest = TailLogsRequest.builder().sourceId("test-source-id").build(); + requestLogsTailLogCacheMultiple(this.logCacheClient, tailRequest); + + this.applications + .logsTail(tailRequest) + .take(3) + .map(e -> e.getLog().getType()) + .as(StepVerifier::create) + .expectNext(LogType.OUT) + .expectNext(LogType.ERR) + .expectNext(LogType.OUT) + .expectComplete() + .verify(Duration.ofSeconds(5)); + } + + @Test + void logsTailLogCacheError() { + TailLogsRequest tailRequest = TailLogsRequest.builder().sourceId("test-source-id").build(); + when(this.logCacheClient.logsTail(tailRequest)) + .thenReturn(Flux.error(new RuntimeException("log-cache unavailable"))); + + this.applications + .logsTail(tailRequest) + .as(StepVerifier::create) + .expectErrorMatches( + t -> + t instanceof RuntimeException + && "log-cache unavailable".equals(t.getMessage())) + .verify(Duration.ofSeconds(5)); + } + + @Test + void logsTailLogCacheOutAndErrEnvelopes() { + TailLogsRequest tailRequest = TailLogsRequest.builder().sourceId("test-source-id").build(); + requestLogsTailLogCacheOutAndErr(this.logCacheClient, tailRequest); + + this.applications + .logsTail(tailRequest) + .take(2) + .as(StepVerifier::create) + .expectNextMatches(e -> LogType.OUT.equals(e.getLog().getType())) + .expectNextMatches(e -> LogType.ERR.equals(e.getLog().getType())) + .expectComplete() + .verify(Duration.ofSeconds(5)); + } + @SuppressWarnings("deprecation") @Test void logsRecentNotSetDoppler() { @@ -5403,6 +5469,90 @@ private static void requestLogsRecentLogCache(LogCacheClient logCacheClient, Str .build())); } + private static void requestLogsTailLogCache( + LogCacheClient logCacheClient, TailLogsRequest tailRequest, String payload) { + when(logCacheClient.logsTail(tailRequest)) + .thenReturn( + Flux.just( + Envelope.builder() + .sourceId(tailRequest.getSourceId()) + .timestamp(System.nanoTime()) + .log( + Log.builder() + .payload(payload) + .type(LogType.OUT) + .build()) + .build())); + } + + /** + * Three envelopes with types OUT, ERR, OUT and strictly ascending timestamps so ordering + * is deterministic. + */ + private static void requestLogsTailLogCacheMultiple( + LogCacheClient logCacheClient, TailLogsRequest tailRequest) { + long base = System.nanoTime(); + when(logCacheClient.logsTail(tailRequest)) + .thenReturn( + Flux.just( + Envelope.builder() + .sourceId(tailRequest.getSourceId()) + .timestamp(base) + .log( + Log.builder() + .payload("msg1") + .type(LogType.OUT) + .build()) + .build(), + Envelope.builder() + .sourceId(tailRequest.getSourceId()) + .timestamp(base + 1) + .log( + Log.builder() + .payload("msg2") + .type(LogType.ERR) + .build()) + .build(), + Envelope.builder() + .sourceId(tailRequest.getSourceId()) + .timestamp(base + 2) + .log( + Log.builder() + .payload("msg3") + .type(LogType.OUT) + .build()) + .build())); + } + + /** + * Two envelopes – one STDOUT, one STDERR – to verify both log types are forwarded. + */ + private static void requestLogsTailLogCacheOutAndErr( + LogCacheClient logCacheClient, TailLogsRequest tailRequest) { + long base = System.nanoTime(); + when(logCacheClient.logsTail(tailRequest)) + .thenReturn( + Flux.just( + Envelope.builder() + .sourceId(tailRequest.getSourceId()) + .timestamp(base) + .log( + Log.builder() + .payload("stdout") + .type(LogType.OUT) + .build()) + .build(), + Envelope.builder() + .sourceId(tailRequest.getSourceId()) + .timestamp(base + 1) + .log( + Log.builder() + .payload("stderr") + .type(LogType.ERR) + .build()) + .build())); + } + private static void requestLogsStream(DopplerClient dopplerClient, String applicationId) { when(dopplerClient.stream(StreamRequest.builder().applicationId(applicationId).build())) .thenReturn( From f7af9613c2cb4f34c2c4a1671af768b3014d1048 Mon Sep 17 00:00:00 2001 From: Pascal Zimmermann Date: Tue, 17 Mar 2026 09:36:49 +0100 Subject: [PATCH 2/6] feat: Add integrationtests --- .../operations/ApplicationsTest.java | 261 +++++++++++++----- 1 file changed, 193 insertions(+), 68 deletions(-) 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 ed8319bafc..e868803eb3 100644 --- a/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java +++ b/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java @@ -25,15 +25,21 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.logging.Level; import org.cloudfoundry.AbstractIntegrationTest; import org.cloudfoundry.CleanupCloudFoundryAfterClass; import org.cloudfoundry.CloudFoundryVersion; import org.cloudfoundry.IfCloudFoundryVersion; -import org.cloudfoundry.RequiresTcpRouting; -import org.cloudfoundry.RequiresV2Api; 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.logcache.v1.TailLogsRequest; import org.cloudfoundry.operations.applications.ApplicationDetail; import org.cloudfoundry.operations.applications.ApplicationEnvironments; import org.cloudfoundry.operations.applications.ApplicationEvent; @@ -82,18 +88,18 @@ import org.cloudfoundry.operations.services.CreateUserProvidedServiceInstanceRequest; import org.cloudfoundry.operations.services.GetServiceInstanceRequest; import org.cloudfoundry.operations.services.ServiceInstance; +import org.cloudfoundry.operations.util.OperationsLogging; import org.cloudfoundry.util.FluentMap; -import org.cloudfoundry.util.PaginationUtils; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.core.io.ClassPathResource; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.publisher.SignalType; import reactor.test.StepVerifier; @CleanupCloudFoundryAfterClass -@RequiresV2Api public final class ApplicationsTest extends AbstractIntegrationTest { private static final String DEFAULT_ROUTER_GROUP = "default-tcp"; @@ -106,6 +112,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. @@ -356,7 +363,6 @@ public void getManifest() throws IOException { } @Test - @RequiresTcpRouting public void getManifestForTcpRoute() throws IOException { String applicationName = this.nameFactory.getApplicationName(); @@ -443,7 +449,6 @@ public void getStopped() throws IOException { } @Test - @RequiresTcpRouting public void getTcp() throws IOException { String applicationName = this.nameFactory.getApplicationName(); String domainName = this.nameFactory.getDomainName(); @@ -507,13 +512,12 @@ public void listTasks() throws IOException { } /** - * Exercise the LogCache client via {@code logs(ApplicationLogsRequest)}. - * LogCache has been a default cf-deployment component since v3.0.0 (July 2018), - * with the {@code /api/v1/read} endpoint available since log-cache-release v2.0.0 - * (October 2018). + * Doppler was dropped in PCF 4.x in favor of logcache. This test does not work + * on TAS 4.x. */ + @Deprecated @Test - @IfCloudFoundryVersion(greaterThanOrEqualTo = CloudFoundryVersion.PCF_2_3) + @IfCloudFoundryVersion(lessThan = CloudFoundryVersion.PCF_4_v2) public void logs() throws IOException { String applicationName = this.nameFactory.getApplicationName(); @@ -538,6 +542,159 @@ public void logs() throws IOException { .verify(Duration.ofMinutes(5)); } + @Test + public void logsRecent() throws IOException { + String applicationName = this.nameFactory.getApplicationName(); + Mono applicationGuid = + getAppGuidFromAppName(cloudFoundryOperations, applicationName); + createApplication( + this.cloudFoundryOperations, + new ClassPathResource("test-application.zip").getFile().toPath(), + applicationName, + false) + .then( + applicationGuid + .map(ApplicationsTest::getReadRequest) + .flatMapMany( + readRequest -> + callLogsRecent( + this.cloudFoundryOperations, + readRequest) + .log(null, Level.ALL, SignalType.ON_NEXT)) + .map(ApplicationsTest::checkOneLogEntry) + .then()) + .as(StepVerifier::create) + .expectComplete() + .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)); + } + + /** + * Integration test for {@link org.cloudfoundry.operations.applications.Applications#logsTail}. + * Verifies that streaming a single LOG envelope from a running application succeeds. + */ + @Test + public void logsTail() 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.cloudFoundryOperations + .applications() + .logsTail( + TailLogsRequest.builder() + .sourceId(appGuid) + .envelopeTypes( + Collections.singletonList( + EnvelopeType.LOG)) + .build()) + .take(1)) + .map(Envelope::getLog) + .map(Log::getType) + .as(StepVerifier::create) + .expectNextMatches( + logType -> LogType.OUT.equals(logType) || LogType.ERR.equals(logType)) + .expectComplete() + .verify(Duration.ofMinutes(5)); + } + + /** + * Integration test for {@link org.cloudfoundry.operations.applications.Applications#logsTail} + * verifying that multiple LOG envelopes can be streamed from a running application. + */ + @Test + public void logsTailMultipleEnvelopes() 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.cloudFoundryOperations + .applications() + .logsTail( + TailLogsRequest.builder() + .sourceId(appGuid) + .envelopeTypes( + Collections.singletonList( + EnvelopeType.LOG)) + .build()) + .take(3)) + .map(Envelope::getLog) + .map(Log::getType) + .as(StepVerifier::create) + .expectNextMatches( + logType -> LogType.OUT.equals(logType) || LogType.ERR.equals(logType)) + .expectNextMatches( + logType -> LogType.OUT.equals(logType) || LogType.ERR.equals(logType)) + .expectNextMatches( + logType -> LogType.OUT.equals(logType) || LogType.ERR.equals(logType)) + .expectComplete() + .verify(Duration.ofMinutes(5)); + } + @Test public void pushBindServices() throws IOException { String applicationName = this.nameFactory.getApplicationName(); @@ -753,59 +910,6 @@ public void pushManifestV3() throws IOException { .verify(Duration.ofMinutes(5)); } - @Test - @IfCloudFoundryVersion(greaterThanOrEqualTo = CloudFoundryVersion.PCF_4_v2) - public void pushManifestV3WithFeature() throws IOException { - String applicationName = this.nameFactory.getApplicationName(); - - final String featureKey = "ssh"; - final boolean featureValue = false; - ManifestV3 manifest = - ManifestV3.builder() - .application( - ManifestV3Application.builder() - .buildpack("staticfile_buildpack") - .disk(512) - .healthCheckType(ApplicationHealthCheck.PORT) - .memory(64) - .name(applicationName) - .feature(featureKey, false) - .path( - new ClassPathResource("test-application.zip") - .getFile() - .toPath()) - .build()) - .build(); - - this.cloudFoundryOperations - .applications() - .pushManifestV3(PushManifestV3Request.builder().manifest(manifest).build()) - .then( - this.cloudFoundryOperations - .applications() - .get(GetApplicationRequest.builder().name(applicationName).build())) - .map(ApplicationDetail::getId) - .flatMapMany( - applicationId -> - PaginationUtils.requestClientV3Resources( - page -> - this.cloudFoundryClient - .applicationsV3() - .listFeatures( - ListApplicationFeaturesRequest - .builder() - .applicationId( - applicationId) - .page(page) - .build()))) - .filter(feature -> featureKey.equals(feature.getName())) - .map(ApplicationFeatureResource::getEnabled) - .as(StepVerifier::create) - .expectNext(featureValue) - .expectComplete() - .verify(Duration.ofMinutes(5)); - } - @Test @IfCloudFoundryVersion(greaterThanOrEqualTo = CloudFoundryVersion.PCF_4_v2) public void pushManifestV3WithMetadata() throws IOException { @@ -1201,7 +1305,6 @@ public void pushRoutePath() throws IOException { } @Test - @RequiresTcpRouting public void pushTcpRoute() throws IOException { String applicationName = this.nameFactory.getApplicationName(); String domainName = this.nameFactory.getDomainName(); @@ -1325,7 +1428,6 @@ public void pushUpdateRoute() throws IOException { } @Test - @RequiresTcpRouting public void pushUpdateTcpRoute() throws IOException { String applicationName = this.nameFactory.getApplicationName(); String domainName = this.nameFactory.getDomainName(); @@ -2196,4 +2298,27 @@ private static Mono requestSshEnabled( .applications() .sshEnabled(ApplicationSshEnabledRequest.builder().name(applicationName).build()); } + + private static ReadRequest getReadRequest(String applicationId) { + return ReadRequest.builder().sourceId(applicationId).build(); + } + + private static Flux callLogsRecent( + CloudFoundryOperations cloudFoundryOperations, ReadRequest readRequest) { + return cloudFoundryOperations.applications().logsRecent(readRequest); + } + + private static Mono getAppGuidFromAppName( + CloudFoundryOperations cloudFoundryOperations, String applicationName) { + return cloudFoundryOperations + .applications() + .get(GetApplicationRequest.builder().name(applicationName).build()) + .map(ApplicationDetail::getId); + } + + private static Log checkOneLogEntry(Log log) { + OperationsLogging.log("one log entry: " + log.getType() + " " + log.getPayloadAsText()); + assertThat(log.getType()).isIn(LogType.OUT, LogType.ERR); + return log; + } } From 712f6d934b14b0755f6a78263f1d0024c693d426 Mon Sep 17 00:00:00 2001 From: I539231 Date: Wed, 10 Jun 2026 07:52:12 +0200 Subject: [PATCH 3/6] fix: Reformat the file Signed-off-by: I539231 --- .../applications/DefaultApplicationsTest.java | 27 ++++++------------- 1 file changed, 8 insertions(+), 19 deletions(-) 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 af61a2363c..24508271df 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 @@ -1371,28 +1371,16 @@ void logsRecentDoppler() { @Test void logsLogCache() { - requestApplications( - this.cloudFoundryClient, - "test-application-name", - TEST_SPACE_ID, - "test-metadata-id"); - requestLogsRecentLogCache(this.logCacheClient, "test-metadata-id"); + ReadRequest readRequest = ReadRequest.builder().sourceId("test-metadata-id").build(); + requestLogsRecentLogCache(this.logCacheClient, readRequest); this.applications - .logs( - ApplicationLogsRequest.builder() - .name("test-application-name") - .recent(true) - .build()) + .logsRecent(readRequest) .as(StepVerifier::create) .expectNextMatches( log -> - log.getMessage().equals("test-payload") - && log.getLogType() == ApplicationLogType.OUT - && log.getSourceId().equals("test-sourceId") - && log.getInstanceId().equals("test-instanceId") - && log.getSourceType().equals("APP/PROC/WEB") - && log.getTimestamp() == 1L) + log.getPayloadAsText().equals("test-payload") + && log.getType() == LogType.OUT) .expectComplete() .verify(Duration.ofSeconds(5)); } @@ -5438,10 +5426,11 @@ private static void requestLogsRecent(DopplerClient dopplerClient, String applic .build())); } - private static void requestLogsRecentLogCache(LogCacheClient logCacheClient, String sourceId) { + private static void requestLogsRecentLogCache( + LogCacheClient logCacheClient, ReadRequest readRequest) { String base64Payload = Base64.getEncoder().encodeToString("test-payload".getBytes(StandardCharsets.UTF_8)); - when(logCacheClient.read(ReadRequest.builder().sourceId(sourceId).build())) + when(logCacheClient.recentLogs(readRequest)) .thenReturn( Mono.just( fill(ReadResponse.builder()) From 4e0562b51b2d180ee0a4c71930b61ccb0d6f239d Mon Sep 17 00:00:00 2001 From: I539231 Date: Tue, 22 Sep 2026 11:24:16 +0200 Subject: [PATCH 4/6] fix: Adjust the error handling, the tests and copyright information --- .../logcache/v1/ReactorLogCacheEndpoints.java | 17 ++++- .../logcache/v1/_TailLogsRequest.java | 2 +- .../operations/ApplicationsTest.java | 63 +++++++++++++++++++ 3 files changed, 80 insertions(+), 2 deletions(-) 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 6df4093ed5..9be2751bd9 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 @@ -21,6 +21,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicLong; +import org.cloudfoundry.AbstractCloudFoundryException; import org.cloudfoundry.logcache.v1.Envelope; import org.cloudfoundry.logcache.v1.EnvelopeType; import org.cloudfoundry.logcache.v1.InfoRequest; @@ -32,11 +33,16 @@ 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; final class ReactorLogCacheEndpoints extends AbstractLogCacheOperations { + private static final Logger LOGGER = + LoggerFactory.getLogger(ReactorLogCacheEndpoints.class); + ReactorLogCacheEndpoints( ConnectionContext connectionContext, Mono root, @@ -116,7 +122,16 @@ Flux logsTail(TailLogsRequest request) { } return read(builder.build()) - .onErrorReturn(ReadResponse.builder().build()) + .onErrorResume( + t -> { + if (t instanceof AbstractCloudFoundryException) { + return Mono.error(t); + } + LOGGER.warn( + "Transient error polling Log Cache, retrying", + t); + return Mono.just(ReadResponse.builder().build()); + }) .flatMapMany( resp -> { List raw = 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 index 1ca32d388a..65208889c5 100644 --- a/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java +++ b/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2021 the original author or authors. + * 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. 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 e868803eb3..f3d658a41f 100644 --- a/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java +++ b/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java @@ -30,7 +30,11 @@ import org.cloudfoundry.CleanupCloudFoundryAfterClass; import org.cloudfoundry.CloudFoundryVersion; import org.cloudfoundry.IfCloudFoundryVersion; +import org.cloudfoundry.RequiresTcpRouting; +import org.cloudfoundry.RequiresV2Api; 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; @@ -90,6 +94,7 @@ import org.cloudfoundry.operations.services.ServiceInstance; import org.cloudfoundry.operations.util.OperationsLogging; import org.cloudfoundry.util.FluentMap; +import org.cloudfoundry.util.PaginationUtils; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -100,6 +105,7 @@ import reactor.test.StepVerifier; @CleanupCloudFoundryAfterClass +@RequiresV2Api public final class ApplicationsTest extends AbstractIntegrationTest { private static final String DEFAULT_ROUTER_GROUP = "default-tcp"; @@ -363,6 +369,7 @@ public void getManifest() throws IOException { } @Test + @RequiresTcpRouting public void getManifestForTcpRoute() throws IOException { String applicationName = this.nameFactory.getApplicationName(); @@ -449,6 +456,7 @@ public void getStopped() throws IOException { } @Test + @RequiresTcpRouting public void getTcp() throws IOException { String applicationName = this.nameFactory.getApplicationName(); String domainName = this.nameFactory.getDomainName(); @@ -910,6 +918,59 @@ public void pushManifestV3() throws IOException { .verify(Duration.ofMinutes(5)); } + @Test + @IfCloudFoundryVersion(greaterThanOrEqualTo = CloudFoundryVersion.PCF_4_v2) + public void pushManifestV3WithFeature() throws IOException { + String applicationName = this.nameFactory.getApplicationName(); + + final String featureKey = "ssh"; + final boolean featureValue = false; + ManifestV3 manifest = + ManifestV3.builder() + .application( + ManifestV3Application.builder() + .buildpack("staticfile_buildpack") + .disk(512) + .healthCheckType(ApplicationHealthCheck.PORT) + .memory(64) + .name(applicationName) + .feature(featureKey, false) + .path( + new ClassPathResource("test-application.zip") + .getFile() + .toPath()) + .build()) + .build(); + + this.cloudFoundryOperations + .applications() + .pushManifestV3(PushManifestV3Request.builder().manifest(manifest).build()) + .then( + this.cloudFoundryOperations + .applications() + .get(GetApplicationRequest.builder().name(applicationName).build())) + .map(ApplicationDetail::getId) + .flatMapMany( + applicationId -> + PaginationUtils.requestClientV3Resources( + page -> + this.cloudFoundryClient + .applicationsV3() + .listFeatures( + ListApplicationFeaturesRequest + .builder() + .applicationId( + applicationId) + .page(page) + .build()))) + .filter(feature -> featureKey.equals(feature.getName())) + .map(ApplicationFeatureResource::getEnabled) + .as(StepVerifier::create) + .expectNext(featureValue) + .expectComplete() + .verify(Duration.ofMinutes(5)); + } + @Test @IfCloudFoundryVersion(greaterThanOrEqualTo = CloudFoundryVersion.PCF_4_v2) public void pushManifestV3WithMetadata() throws IOException { @@ -1200,6 +1261,7 @@ public void pushNoRoute() throws IOException { } @Test + @RequiresTcpRouting public void pushPrivateDomain() throws IOException { String applicationName = this.nameFactory.getApplicationName(); String domainName = this.nameFactory.getDomainName(); @@ -1428,6 +1490,7 @@ public void pushUpdateRoute() throws IOException { } @Test + @RequiresTcpRouting public void pushUpdateTcpRoute() throws IOException { String applicationName = this.nameFactory.getApplicationName(); String domainName = this.nameFactory.getDomainName(); From 47b07d95e2e8910dcca308d4cba5013b04a9679b Mon Sep 17 00:00:00 2001 From: Pascal Zimmermann Date: Sat, 17 Oct 2026 10:59:26 +0200 Subject: [PATCH 5/6] fix: Remove the recent logs part --- .../logcache/v1/ReactorLogCacheEndpoints.java | 4 - .../logcache/v1/_ReactorLogCacheClient.java | 5 - .../logcache/v1/LogCacheClient.java | 8 -- .../operations/applications/Applications.java | 16 +--- .../applications/DefaultApplications.java | 95 +++++++++++-------- .../applications/DefaultApplicationsTest.java | 27 ++++-- .../operations/ApplicationsTest.java | 61 +----------- 7 files changed, 85 insertions(+), 131 deletions(-) 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 9be2751bd9..747d8cd571 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 @@ -63,10 +63,6 @@ Mono read(ReadRequest request) { return get(request, ReadResponse.class, "read", request.getSourceId()).checkpoint(); } - Mono recentLogs(ReadRequest request) { - return read(request); - } - /** * Continuously polls Log Cache and emits new {@link Envelope}s as they arrive. * 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 0236da2188..7f4938f1f7 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 @@ -56,11 +56,6 @@ public Mono read(ReadRequest request) { return getReactorLogCacheEndpoints().read(request); } - @Override - public Mono recentLogs(ReadRequest request) { - return getReactorLogCacheEndpoints().recentLogs(request); - } - @Override public Flux logsTail(TailLogsRequest request) { return getReactorLogCacheEndpoints().logsTail(request); 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 6b91461fe3..82466dfdf3 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 @@ -48,14 +48,6 @@ public interface LogCacheClient { */ Mono read(ReadRequest request); - /** - * Makes the Log Cache RecentLogs /api/v1/read request - * - * @param request the Recent Logs request - * @return the events from the recent logs - */ - Mono recentLogs(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 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 5d95db5d01..0207c46a7f 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 @@ -18,8 +18,6 @@ import org.cloudfoundry.doppler.LogMessage; import org.cloudfoundry.logcache.v1.Envelope; -import org.cloudfoundry.logcache.v1.Log; -import org.cloudfoundry.logcache.v1.ReadRequest; import org.cloudfoundry.logcache.v1.TailLogsRequest; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -130,15 +128,6 @@ public interface Applications { @Deprecated Flux logs(LogsRequest request); - /** - * List the applications logs from logCacheClient. - * If no messages are available, an empty Flux is returned. - * - * @param request the application logs request - * @return the applications logs - */ - Flux logsRecent(ReadRequest request); - /** * Continuously streams application log envelopes from Log Cache by repeatedly polling * the {@code /api/v1/read} endpoint. The returned {@link Flux} is infinite – cancel it @@ -151,8 +140,9 @@ public interface Applications { /** * List the applications logs. - * Only works with {@code Loggregator < 107.0}, shipped in {@code CFD < 24.3} - * and {@code TAS < 4.0}. + * 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. * * @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 23d1fb70cd..6ccc95a22e 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,7 +155,6 @@ import org.cloudfoundry.doppler.RecentLogsRequest; import org.cloudfoundry.doppler.StreamRequest; import org.cloudfoundry.logcache.v1.EnvelopeBatch; -import org.cloudfoundry.logcache.v1.Log; import org.cloudfoundry.logcache.v1.LogCacheClient; import org.cloudfoundry.logcache.v1.ReadRequest; import org.cloudfoundry.logcache.v1.TailLogsRequest; @@ -559,13 +558,6 @@ public Flux logs(LogsRequest request) { .checkpoint(); } - @Override - public Flux logsRecent(ReadRequest request) { - return getRecentLogsLogCache(this.logCacheClient, request) - .transform(OperationsLogging.log("Get Application Logs")) - .checkpoint(); - } - @Override public Flux logsTail(TailLogsRequest request) { return this.logCacheClient @@ -576,22 +568,23 @@ public Flux logsTail(TailLogsRequest requ @Override public Flux logs(ApplicationLogsRequest request) { - return logs(LogsRequest.builder() - .name(request.getName()) - .recent(request.getRecent()) - .build()) - .map( - logMessage -> - ApplicationLog.builder() - .sourceId(logMessage.getApplicationId()) - .sourceType(logMessage.getSourceType()) - .instanceId(logMessage.getSourceInstance()) - .message(logMessage.getMessage()) - .timestamp(logMessage.getTimestamp()) - .logType( - ApplicationLogType.from( - logMessage.getMessageType().name())) - .build()); + 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)) + .transform(OperationsLogging.log("Get Application Logs")) + .checkpoint(); + } else { + return logs(LogsRequest.builder().name(request.getName()).recent(false).build()) + .map(DefaultApplications::toApplicationLog); + } } @Override @@ -1646,15 +1639,30 @@ private static Flux getLogs( } } - private static Flux getRecentLogsLogCache( - Mono logCacheClient, ReadRequest readRequest) { - return requestLogsRecentLogCache(logCacheClient, readRequest) - .map(EnvelopeBatch::getBatch) - .map(List::stream) - .flatMapIterable(envelopeStream -> envelopeStream.collect(Collectors.toList())) + private static Flux getLogsLogCache( + Mono logCacheClient, String applicationId) { + return requestLogsRecentLogCache(logCacheClient, applicationId) .filter(e -> e.getLog() != null) .sort(LOG_MESSAGE_COMPARATOR_LOG_CACHE) - .map(org.cloudfoundry.logcache.v1.Envelope::getLog); + .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") @@ -2550,12 +2558,14 @@ private static Flux requestLogsRecent( RecentLogsRequest.builder().applicationId(applicationId).build())); } - private static Mono requestLogsRecentLogCache( - Mono logCacheClient, ReadRequest readRequest) { - return logCacheClient.flatMap( - client -> - client.recentLogs(readRequest) - .flatMap(response -> Mono.justOrEmpty(response.getEnvelopes()))); + private static Flux requestLogsRecentLogCache( + Mono logCacheClient, String applicationId) { + return logCacheClient + .flatMap( + client -> + client.read(ReadRequest.builder().sourceId(applicationId).build())) + .flatMap(response -> Mono.justOrEmpty(response.getEnvelopes())) + .flatMapIterable(EnvelopeBatch::getBatch); } private static Flux requestLogsStream( @@ -2963,6 +2973,17 @@ 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 24508271df..af61a2363c 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 @@ -1371,16 +1371,28 @@ void logsRecentDoppler() { @Test void logsLogCache() { - ReadRequest readRequest = ReadRequest.builder().sourceId("test-metadata-id").build(); - requestLogsRecentLogCache(this.logCacheClient, readRequest); + requestApplications( + this.cloudFoundryClient, + "test-application-name", + TEST_SPACE_ID, + "test-metadata-id"); + requestLogsRecentLogCache(this.logCacheClient, "test-metadata-id"); this.applications - .logsRecent(readRequest) + .logs( + ApplicationLogsRequest.builder() + .name("test-application-name") + .recent(true) + .build()) .as(StepVerifier::create) .expectNextMatches( log -> - log.getPayloadAsText().equals("test-payload") - && log.getType() == LogType.OUT) + log.getMessage().equals("test-payload") + && log.getLogType() == ApplicationLogType.OUT + && log.getSourceId().equals("test-sourceId") + && log.getInstanceId().equals("test-instanceId") + && log.getSourceType().equals("APP/PROC/WEB") + && log.getTimestamp() == 1L) .expectComplete() .verify(Duration.ofSeconds(5)); } @@ -5426,11 +5438,10 @@ private static void requestLogsRecent(DopplerClient dopplerClient, String applic .build())); } - private static void requestLogsRecentLogCache( - LogCacheClient logCacheClient, ReadRequest readRequest) { + private static void requestLogsRecentLogCache(LogCacheClient logCacheClient, String sourceId) { String base64Payload = Base64.getEncoder().encodeToString("test-payload".getBytes(StandardCharsets.UTF_8)); - when(logCacheClient.recentLogs(readRequest)) + when(logCacheClient.read(ReadRequest.builder().sourceId(sourceId).build())) .thenReturn( Mono.just( fill(ReadResponse.builder()) 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 f3d658a41f..45f1f63ac6 100644 --- a/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java +++ b/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java @@ -25,7 +25,6 @@ import java.util.Collections; import java.util.List; import java.util.Map; -import java.util.logging.Level; import org.cloudfoundry.AbstractIntegrationTest; import org.cloudfoundry.CleanupCloudFoundryAfterClass; import org.cloudfoundry.CloudFoundryVersion; @@ -92,7 +91,6 @@ import org.cloudfoundry.operations.services.CreateUserProvidedServiceInstanceRequest; import org.cloudfoundry.operations.services.GetServiceInstanceRequest; import org.cloudfoundry.operations.services.ServiceInstance; -import org.cloudfoundry.operations.util.OperationsLogging; import org.cloudfoundry.util.FluentMap; import org.cloudfoundry.util.PaginationUtils; import org.junit.jupiter.api.Test; @@ -101,7 +99,6 @@ import org.springframework.core.io.ClassPathResource; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.core.publisher.SignalType; import reactor.test.StepVerifier; @CleanupCloudFoundryAfterClass @@ -520,12 +517,13 @@ public void listTasks() throws IOException { } /** - * Doppler was dropped in PCF 4.x in favor of logcache. This test does not work - * on TAS 4.x. + * Exercise the LogCache client via {@code logs(ApplicationLogsRequest)}. + * LogCache has been a default cf-deployment component since v3.0.0 (July 2018), + * with the {@code /api/v1/read} endpoint available since log-cache-release v2.0.0 + * (October 2018). */ - @Deprecated @Test - @IfCloudFoundryVersion(lessThan = CloudFoundryVersion.PCF_4_v2) + @IfCloudFoundryVersion(greaterThanOrEqualTo = CloudFoundryVersion.PCF_2_3) public void logs() throws IOException { String applicationName = this.nameFactory.getApplicationName(); @@ -550,32 +548,6 @@ public void logs() throws IOException { .verify(Duration.ofMinutes(5)); } - @Test - public void logsRecent() throws IOException { - String applicationName = this.nameFactory.getApplicationName(); - Mono applicationGuid = - getAppGuidFromAppName(cloudFoundryOperations, applicationName); - createApplication( - this.cloudFoundryOperations, - new ClassPathResource("test-application.zip").getFile().toPath(), - applicationName, - false) - .then( - applicationGuid - .map(ApplicationsTest::getReadRequest) - .flatMapMany( - readRequest -> - callLogsRecent( - this.cloudFoundryOperations, - readRequest) - .log(null, Level.ALL, SignalType.ON_NEXT)) - .map(ApplicationsTest::checkOneLogEntry) - .then()) - .as(StepVerifier::create) - .expectComplete() - .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 @@ -2361,27 +2333,4 @@ private static Mono requestSshEnabled( .applications() .sshEnabled(ApplicationSshEnabledRequest.builder().name(applicationName).build()); } - - private static ReadRequest getReadRequest(String applicationId) { - return ReadRequest.builder().sourceId(applicationId).build(); - } - - private static Flux callLogsRecent( - CloudFoundryOperations cloudFoundryOperations, ReadRequest readRequest) { - return cloudFoundryOperations.applications().logsRecent(readRequest); - } - - private static Mono getAppGuidFromAppName( - CloudFoundryOperations cloudFoundryOperations, String applicationName) { - return cloudFoundryOperations - .applications() - .get(GetApplicationRequest.builder().name(applicationName).build()) - .map(ApplicationDetail::getId); - } - - private static Log checkOneLogEntry(Log log) { - OperationsLogging.log("one log entry: " + log.getType() + " " + log.getPayloadAsText()); - assertThat(log.getType()).isIn(LogType.OUT, LogType.ERR); - return log; - } } From c0ec0219fe0e11b7ca37ca0eb93c3584f636f77d Mon Sep 17 00:00:00 2001 From: Pascal Zimmermann Date: Sat, 17 Oct 2026 13:28:43 +0200 Subject: [PATCH 6/6] feat: Update the implementation --- .../logcache/v1/ReactorLogCacheEndpoints.java | 206 +++++++++--------- .../reactor/util/ErrorPayloadMappers.java | 3 + .../v1/ReactorLogCacheClientTest.java | 95 ++++++++ .../GET_read_tail_page1_response.json | 22 ++ .../GET_read_tail_page2_response.json | 14 ++ .../logcache/v1/_TailLogsRequest.java | 11 + .../operations/applications/Applications.java | 18 +- .../applications/DefaultApplications.java | 80 ++++--- .../applications/DefaultApplicationsTest.java | 185 ++++++---------- .../operations/ApplicationsTest.java | 82 +------ 10 files changed, 382 insertions(+), 334 deletions(-) create mode 100644 cloudfoundry-client-reactor/src/test/resources/fixtures/logcache.v1/GET_read_tail_page1_response.json create mode 100644 cloudfoundry-client-reactor/src/test/resources/fixtures/logcache.v1/GET_read_tail_page2_response.json 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 747d8cd571..71dc04e868 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,11 +16,13 @@ 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 java.util.concurrent.atomic.AtomicLong; import org.cloudfoundry.AbstractCloudFoundryException; import org.cloudfoundry.logcache.v1.Envelope; import org.cloudfoundry.logcache.v1.EnvelopeType; @@ -37,11 +39,15 @@ 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 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, @@ -66,112 +72,112 @@ Mono read(ReadRequest request) { /** * Continuously polls Log Cache and emits new {@link Envelope}s as they arrive. * - *

Mirrors the Go {@code logcache.Walk()} / {@code cf tail --follow} semantics: + *

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

    - *
  1. Start the cursor at {@code startTime} (defaults to now − 5 s in - * nanoseconds).
    2. + *
  2. 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.
    3. *
  3. Issue {@code GET /api/v1/read/{sourceId}?start_time=cursor}.
    4. - *
  4. Emit every returned envelope in ascending timestamp order and advance - * the cursor to {@code lastTimestamp + 1}.
    5. - *
  5. When the batch is empty, wait {@code pollInterval} before the next poll.
    6. - *
  6. Repeat forever – the caller cancels the subscription to stop.
    7. + *
  7. 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}.
    8. + *
  8. When nothing can be emitted, wait {@code pollInterval} before polling again.
    9. + *
  9. 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.
    10. *
- * Fully non-blocking: no {@code Thread.sleep}. + * The stream never completes on its own – the caller cancels the subscription to stop. */ Flux logsTail(TailLogsRequest request) { - long defaultStartNanos = (System.currentTimeMillis() - 5_000L) * 1_000_000L; - AtomicLong cursor = - new AtomicLong( - request.getStartTime() != null - ? request.getStartTime() - : defaultStartNanos); - List envelopeTypes = request.getEnvelopeTypes() != null ? request.getEnvelopeTypes() : Collections.emptyList(); - String nameFilter = request.getNameFilter(); - - /* - * Strategy (mirrors Go's logcache.Walk): - * – Mono.defer builds a fresh ReadRequest from the mutable cursor on every repetition. - * – The Mono returns either the sorted batch (non-empty) or an empty list. - * – flatMapMany turns each batch into a stream of individual Envelope items. - * – repeat() subscribes again after each completion. - * – When the batch was empty we insert a delay via Mono.delay before the next - * repetition so we do not hammer the server. We signal "empty" by returning - * a sentinel Mono (false = was empty, true = had data) and use - * repeatWhen to conditionally delay. - */ - return Flux.defer( - () -> { - // Build the read request from the current cursor position. - ReadRequest.Builder builder = - ReadRequest.builder() - .sourceId(request.getSourceId()) - .startTime(cursor.get()); - if (!envelopeTypes.isEmpty()) { - builder.envelopeTypes(envelopeTypes); - } - if (nameFilter != null && !nameFilter.isEmpty()) { - builder.nameFilter(nameFilter); - } - - return read(builder.build()) - .onErrorResume( - t -> { - if (t instanceof AbstractCloudFoundryException) { - return Mono.error(t); - } + + 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, retrying", - t); - return Mono.just(ReadResponse.builder().build()); - }) - .flatMapMany( - resp -> { - List raw = - resp.getEnvelopes() != null - ? resp.getEnvelopes().getBatch() - : Collections.emptyList(); - - if (raw.isEmpty()) { - // Signal "no data" so repeatWhen can insert the - // back-off delay. - return Flux.empty(); - } - - // Sort ascending by timestamp and advance the - // cursor. - List sorted = new ArrayList<>(raw); - sorted.sort( - (a, b) -> - Long.compare( - a.getTimestamp() != null - ? a.getTimestamp() - : 0L, - b.getTimestamp() != null - ? b.getTimestamp() - : 0L)); - - Envelope last = sorted.get(sorted.size() - 1); - cursor.set( - (last.getTimestamp() != null - ? last.getTimestamp() - : cursor.get()) - + 1); - - return Flux.fromIterable(sorted); - }); - }) - // repeatWhen receives a Flux where each element is the count of items - // emitted in the previous cycle (0 = empty batch → insert delay). - .repeatWhen( - companion -> - companion.flatMap( - count -> - count == 0 - ? Mono.delay(request.getPollInterval()) - : Mono.just(count))); + "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/util/ErrorPayloadMappers.java b/cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/util/ErrorPayloadMappers.java index 282b5ec2e6..4519ca5560 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 740461dc46..abe5181d6f 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 0000000000..cf925863e6 --- /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 0000000000..b4bca11a1b --- /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/_TailLogsRequest.java b/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java index 65208889c5..34dd812540 100644 --- a/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java +++ b/cloudfoundry-client/src/main/java/org/cloudfoundry/logcache/v1/_TailLogsRequest.java @@ -61,5 +61,16 @@ abstract class _TailLogsRequest { 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 0207c46a7f..d15df7f231 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 @@ -17,8 +17,6 @@ package org.cloudfoundry.operations.applications; import org.cloudfoundry.doppler.LogMessage; -import org.cloudfoundry.logcache.v1.Envelope; -import org.cloudfoundry.logcache.v1.TailLogsRequest; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -128,21 +126,11 @@ public interface Applications { @Deprecated Flux logs(LogsRequest request); - /** - * Continuously streams application log envelopes from Log Cache by repeatedly polling - * the {@code /api/v1/read} endpoint. The returned {@link Flux} is infinite – cancel it - * to stop streaming. This is the Java equivalent of {@code cf tail --follow}. - * - * @param request the tail request (source id, optional filters, poll interval) - * @return an infinite stream of envelopes - */ - Flux logsTail(TailLogsRequest request); - /** * 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 6ccc95a22e..ec4e7a0d6d 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,6 +155,7 @@ 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; @@ -558,32 +559,27 @@ public Flux logs(LogsRequest request) { .checkpoint(); } - @Override - public Flux logsTail(TailLogsRequest request) { - return this.logCacheClient - .flatMapMany(client -> client.logsTail(request)) - .transform(OperationsLogging.log("Tail Application Logs")) - .checkpoint(); - } - @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(); } } @@ -1665,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 = @@ -2973,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 af61a2363c..3fbffef118 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,6 +147,7 @@ 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; @@ -1398,47 +1399,54 @@ void logsLogCache() { } @Test - void logsTailLogCache() { - TailLogsRequest tailRequest = TailLogsRequest.builder().sourceId("test-source-id").build(); - requestLogsTailLogCache(this.logCacheClient, tailRequest, "test-tail-payload"); - - this.applications - .logsTail(tailRequest) - .take(1) - .as(StepVerifier::create) - .expectNextMatches( - envelope -> - envelope.getLog() != null - && LogType.OUT.equals(envelope.getLog().getType())) - .expectComplete() - .verify(Duration.ofSeconds(5)); - } - - @Test - void logsTailLogCacheMultipleEnvelopes() { - TailLogsRequest tailRequest = TailLogsRequest.builder().sourceId("test-source-id").build(); - requestLogsTailLogCacheMultiple(this.logCacheClient, tailRequest); + 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 - .logsTail(tailRequest) + .logs(ApplicationLogsRequest.builder().name("test-application-name").build()) .take(3) - .map(e -> e.getLog().getType()) .as(StepVerifier::create) - .expectNext(LogType.OUT) - .expectNext(LogType.ERR) - .expectNext(LogType.OUT) + .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 logsTailLogCacheError() { - TailLogsRequest tailRequest = TailLogsRequest.builder().sourceId("test-source-id").build(); - when(this.logCacheClient.logsTail(tailRequest)) + 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 - .logsTail(tailRequest) + .logs( + ApplicationLogsRequest.builder() + .name("test-application-name") + .recent(false) + .build()) .as(StepVerifier::create) .expectErrorMatches( t -> @@ -1448,17 +1456,18 @@ void logsTailLogCacheError() { } @Test - void logsTailLogCacheOutAndErrEnvelopes() { - TailLogsRequest tailRequest = TailLogsRequest.builder().sourceId("test-source-id").build(); - requestLogsTailLogCacheOutAndErr(this.logCacheClient, tailRequest); + void logsStreamLogCacheApplicationDoesNotExist() { + requestApplicationsEmpty(this.cloudFoundryClient, "test-application-name", TEST_SPACE_ID); this.applications - .logsTail(tailRequest) - .take(2) + .logs(ApplicationLogsRequest.builder().name("test-application-name").build()) .as(StepVerifier::create) - .expectNextMatches(e -> LogType.OUT.equals(e.getLog().getType())) - .expectNextMatches(e -> LogType.ERR.equals(e.getLog().getType())) - .expectComplete() + .consumeErrorWith( + t -> + assertThat(t) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Application test-application-name does not exist")) .verify(Duration.ofSeconds(5)); } @@ -5469,88 +5478,32 @@ private static void requestLogsRecentLogCache(LogCacheClient logCacheClient, Str .build())); } - private static void requestLogsTailLogCache( - LogCacheClient logCacheClient, TailLogsRequest tailRequest, String payload) { - when(logCacheClient.logsTail(tailRequest)) - .thenReturn( - Flux.just( - Envelope.builder() - .sourceId(tailRequest.getSourceId()) - .timestamp(System.nanoTime()) - .log( - Log.builder() - .payload(payload) - .type(LogType.OUT) - .build()) - .build())); - } - - /** - * Three envelopes with types OUT, ERR, OUT and strictly ascending timestamps so ordering - * is deterministic. - */ - private static void requestLogsTailLogCacheMultiple( - LogCacheClient logCacheClient, TailLogsRequest tailRequest) { - long base = System.nanoTime(); - when(logCacheClient.logsTail(tailRequest)) - .thenReturn( - Flux.just( - Envelope.builder() - .sourceId(tailRequest.getSourceId()) - .timestamp(base) - .log( - Log.builder() - .payload("msg1") - .type(LogType.OUT) - .build()) - .build(), - Envelope.builder() - .sourceId(tailRequest.getSourceId()) - .timestamp(base + 1) - .log( - Log.builder() - .payload("msg2") - .type(LogType.ERR) - .build()) - .build(), - Envelope.builder() - .sourceId(tailRequest.getSourceId()) - .timestamp(base + 2) - .log( - Log.builder() - .payload("msg3") - .type(LogType.OUT) - .build()) - .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(); } - /** - * Two envelopes – one STDOUT, one STDERR – to verify both log types are forwarded. - */ - private static void requestLogsTailLogCacheOutAndErr( - LogCacheClient logCacheClient, TailLogsRequest tailRequest) { - long base = System.nanoTime(); + private static void requestLogsTailLogCache( + LogCacheClient logCacheClient, TailLogsRequest tailRequest, Envelope... envelopes) { when(logCacheClient.logsTail(tailRequest)) - .thenReturn( - Flux.just( - Envelope.builder() - .sourceId(tailRequest.getSourceId()) - .timestamp(base) - .log( - Log.builder() - .payload("stdout") - .type(LogType.OUT) - .build()) - .build(), - Envelope.builder() - .sourceId(tailRequest.getSourceId()) - .timestamp(base + 1) - .log( - Log.builder() - .payload("stderr") - .type(LogType.ERR) - .build()) - .build())); + .thenReturn(Flux.concat(Flux.just(envelopes), Flux.never())); } private static void requestLogsStream(DopplerClient dopplerClient, String applicationId) { 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 45f1f63ac6..7238e9651e 100644 --- a/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java +++ b/integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java @@ -42,7 +42,6 @@ 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.applications.ApplicationDetail; import org.cloudfoundry.operations.applications.ApplicationEnvironments; import org.cloudfoundry.operations.applications.ApplicationEvent; @@ -588,11 +587,10 @@ public void logCacheLogs() throws IOException { } /** - * Integration test for {@link org.cloudfoundry.operations.applications.Applications#logsTail}. - * Verifies that streaming a single LOG envelope from a running application succeeds. + * Streams application logs via {@code logs(ApplicationLogsRequest)} (Log Cache tailing). */ @Test - public void logsTail() throws IOException { + public void logsStream() throws IOException { String applicationName = this.nameFactory.getApplicationName(); createApplication( @@ -600,77 +598,17 @@ public void logsTail() throws IOException { 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.cloudFoundryOperations - .applications() - .logsTail( - TailLogsRequest.builder() - .sourceId(appGuid) - .envelopeTypes( - Collections.singletonList( - EnvelopeType.LOG)) - .build()) - .take(1)) - .map(Envelope::getLog) - .map(Log::getType) - .as(StepVerifier::create) - .expectNextMatches( - logType -> LogType.OUT.equals(logType) || LogType.ERR.equals(logType)) - .expectComplete() - .verify(Duration.ofMinutes(5)); - } - - /** - * Integration test for {@link org.cloudfoundry.operations.applications.Applications#logsTail} - * verifying that multiple LOG envelopes can be streamed from a running application. - */ - @Test - public void logsTailMultipleEnvelopes() throws IOException { - String applicationName = this.nameFactory.getApplicationName(); - - createApplication( - this.cloudFoundryOperations, - new ClassPathResource("test-application.zip").getFile().toPath(), - applicationName, - false) - .then( + .thenMany( this.cloudFoundryOperations .applications() - .get( - GetApplicationRequest.builder() + .logs( + ApplicationLogsRequest.builder() .name(applicationName) - .build())) - .map(ApplicationDetail::getId) - .flatMapMany( - appGuid -> - this.cloudFoundryOperations - .applications() - .logsTail( - TailLogsRequest.builder() - .sourceId(appGuid) - .envelopeTypes( - Collections.singletonList( - EnvelopeType.LOG)) - .build()) - .take(3)) - .map(Envelope::getLog) - .map(Log::getType) + .build()) + .take(3)) + .map(ApplicationLog::getLogType) .as(StepVerifier::create) - .expectNextMatches( - logType -> LogType.OUT.equals(logType) || LogType.ERR.equals(logType)) - .expectNextMatches( - logType -> LogType.OUT.equals(logType) || LogType.ERR.equals(logType)) - .expectNextMatches( - logType -> LogType.OUT.equals(logType) || LogType.ERR.equals(logType)) + .expectNextCount(3) .expectComplete() .verify(Duration.ofMinutes(5)); } @@ -1233,7 +1171,6 @@ public void pushNoRoute() throws IOException { } @Test - @RequiresTcpRouting public void pushPrivateDomain() throws IOException { String applicationName = this.nameFactory.getApplicationName(); String domainName = this.nameFactory.getDomainName(); @@ -1339,6 +1276,7 @@ public void pushRoutePath() throws IOException { } @Test + @RequiresTcpRouting public void pushTcpRoute() throws IOException { String applicationName = this.nameFactory.getApplicationName(); String domainName = this.nameFactory.getDomainName();