logcache client impl

Author: David O'Sullivan

cloudfoundry-client-reactor/src/test/java/org/cloudfoundry/reactor/client/v3/serviceInstances/ReactorServiceInstancesV3Test.java

@@ -62,6 +62,7 @@
 import org.cloudfoundry.reactor.TestRequest;
 import org.cloudfoundry.reactor.TestResponse;
 import org.cloudfoundry.reactor.client.AbstractClientApiTest;
+import org.cloudfoundry.reactor.client.v3.serviceinstances.ReactorServiceInstancesV3;
 import org.junit.jupiter.api.Test;
 import reactor.test.StepVerifier;
 

cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/_DefaultCloudFoundryOperations.java

@@ -179,19 +179,19 @@ Mono<CloudFoundryClient> getCloudFoundryClientPublisher() {
     @Nullable
     abstract DopplerClient getDopplerClient();
 
-        /**
-     * The {@link LogCacheClient} to use for operations functionality
-     */
-    @Nullable
-    abstract LogCacheClient getLogCacheClient();
-
     @Value.Derived
     Mono<DopplerClient> getDopplerClientPublisher() {
         return Optional.ofNullable(getDopplerClient())
             .map(Mono::just)
             .orElse(Mono.error(new IllegalStateException("DopplerClient must be set")));
     }
 
+    /**
+     * The {@link LogCacheClient} to use for operations functionality
+     */
+    @Nullable
+    abstract LogCacheClient getLogCacheClient();
+
     @Value.Derived
     Mono<LogCacheClient> getLogCacheClientPublisher() {
         return Optional.ofNullable(getLogCacheClient())

cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/Applications.java

@@ -18,6 +18,8 @@
 
 import org.cloudfoundry.doppler.LogMessage;
 import org.cloudfoundry.logcache.v1.Log;
+import org.cloudfoundry.logcache.v1.ReadRequest;
+import org.cloudfoundry.logcache.v1.ReadResponse;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -122,7 +124,7 @@ public interface Applications {
      * @param request the application logs request
      * @return the applications logs
      */
-    Flux<Log> logs(LogsRequest request);
+    Flux<Log> logs(ReadRequest request);
 
     /**
      * Push a specific application

cloudfoundry-operations/src/main/java/org/cloudfoundry/operations/applications/DefaultApplications.java

@@ -159,6 +159,7 @@
 import org.cloudfoundry.logcache.v1.Log;
 import org.cloudfoundry.logcache.v1.LogCacheClient;
 import org.cloudfoundry.logcache.v1.ReadRequest;
+import org.cloudfoundry.logcache.v1.ReadResponse;
 import org.cloudfoundry.operations.util.OperationsLogging;
 import org.cloudfoundry.util.DateUtils;
 import org.cloudfoundry.util.DelayTimeoutException;
@@ -541,13 +542,13 @@ public Flux<Task> listTasks(ListApplicationTasksRequest request) {
     }
 
     @Override
-    public Flux<Log> logs(LogsRequest request) {
+    public Flux<Log> logs(ReadRequest request) {
         return Mono.zip(this.cloudFoundryClient, this.spaceId)
                 .flatMap(
                         function(
                                 (cloudFoundryClient, spaceId) ->
                                         getApplicationId(
-                                                cloudFoundryClient, request.getName(), spaceId)))
+                                                cloudFoundryClient, request.getSourceId(), spaceId)))
                 .flatMapMany(
                         applicationId ->
                                 getRecentLogs(this.logCacheClient, applicationId))
@@ -664,7 +665,6 @@ public Mono<Void> pushManifestV3(PushManifestV3Request request) {
         } catch (IOException e) {
             throw new RuntimeException("Could not serialize manifest", e);
         }
-
         return Mono.zip(this.cloudFoundryClient, this.spaceId)
                 .flatMap(
                         function(
@@ -1582,30 +1582,29 @@ private static int getInstances(AbstractApplicationResource resource) {
                 .orElse(0);
     }
 
-   /*  private static Flux<Log> getLogs(
-            Mono<LogCacheClient> logCacheClient, String applicationId, Boolean recent) {
+    private static Flux<LogMessage> getLogs(
+            Mono<DopplerClient> dopplerClient, String applicationId, Boolean recent) {
         if (Optional.ofNullable(recent).orElse(false)) {
-            return getRecentLogs(logCacheClient, applicationId);
+            return requestLogsRecent(dopplerClient, applicationId)
+                    .filter(e -> EventType.LOG_MESSAGE == e.getEventType())
+                    .map(Envelope::getLogMessage)
+                    .collectSortedList(LOG_MESSAGE_COMPARATOR)
+                    .flatMapIterable(d -> d);
+        } else {
+            return requestLogsStream(dopplerClient, applicationId)
+                    .filter(e -> EventType.LOG_MESSAGE == e.getEventType())
+                    .map(Envelope::getLogMessage)
+                    .transformDeferred(
+                            SortingUtils.timespan(LOG_MESSAGE_COMPARATOR, LOG_MESSAGE_TIMESPAN));
         }
-    }*/
+    }
 
     private static Flux<Log> getRecentLogs(Mono<LogCacheClient> logCacheClient, String applicationId) {
         return requestLogsRecentLogCache(logCacheClient, applicationId)
-                .filter(e -> EnvelopeType.LOG.getValue().equals(e.getLog().getType().getValue()))
-              //  .collectSortedList(LOG_MESSAGE_COMPARATOR_LOG_CACHE)
                 .sort(LOG_MESSAGE_COMPARATOR_LOG_CACHE)
                 .map(org.cloudfoundry.logcache.v1.Envelope::getLog);
     }
 
-/*     private static Flux<org.cloudfoundry.logcache.v1.Log> getRecentLogs(Mono<LogCacheClient> logCacheClient, String applicationId) {
-        return requestLogsRecentLogCache(logCacheClient, applicationId)
-                .filter(e -> EnvelopeType.LOG.getValue().equals(e.getLog().getType().getValue()))
-                .sort(LOG_MESSAGE_COMPARATOR_LOG_CACHE)
-                .map(org.cloudfoundry.logcache.v1.Envelope::getLog)
-                .collectList()
-                .flatMapIterable(d1 -> d1).cast(org.cloudfoundry.logcache.v1.Log.class);
-    } */
-
     @SuppressWarnings("unchecked")
     private static Map<String, Object> getMetadataRequest(EventEntity entity) {
         Map<String, Optional<Object>> metadata =

cloudfoundry-operations/src/test/java/org/cloudfoundry/operations/applications/DefaultApplicationsTest.java

@@ -144,8 +144,11 @@
 import org.cloudfoundry.doppler.LogMessage;
 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.Log;
 import org.cloudfoundry.logcache.v1.LogCacheClient;
+import org.cloudfoundry.logcache.v1.LogType;
 import org.cloudfoundry.logcache.v1.ReadRequest;
 import org.cloudfoundry.logcache.v1.ReadResponse;
 import org.cloudfoundry.operations.AbstractOperationsTest;
@@ -1318,10 +1321,10 @@ void logs() {
                 "test-application-name",
                 TEST_SPACE_ID,
                 "test-metadata-id");
-        requestLogsStream(this.dopplerClient, "test-metadata-id");
+        requestLogsRecentLogCache(this.logCacheClient, "test-application-name");
 
         this.applications
-                .logs(LogsRequest.builder().name("test-application-name").recent(false).build())
+                .logs(ReadRequest.builder().sourceId("test-application-name").build())
                 .as(StepVerifier::create)
                 .expectNext(fill(Log.builder(), "log-message-").build())
                 .expectComplete()
@@ -1333,7 +1336,7 @@ void logsNoApp() {
         requestApplicationsEmpty(this.cloudFoundryClient, "test-application-name", TEST_SPACE_ID);
 
         this.applications
-                .logs(LogsRequest.builder().name("test-application-name").build())
+                .logs(ReadRequest.builder().sourceId("test-application-name").build())
                 .as(StepVerifier::create)
                 .consumeErrorWith(
                         t ->
@@ -1354,7 +1357,7 @@ void logsRecent() {
         requestLogsRecentLogCache(this.logCacheClient, "test-metadata-id");
 
         this.applications
-                .logs(LogsRequest.builder().name("test-application-name").recent(true).build())
+                .logs(ReadRequest.builder().sourceId("test-application-name").build())
                 .as(StepVerifier::create)
                 .expectNext(fill(Log.builder(), "log-message-").build())
                 .expectComplete()
@@ -1371,7 +1374,7 @@ void logsRecentNotSet() {
         requestLogsStream(this.dopplerClient, "test-metadata-id");
 
         this.applications
-                .logs(LogsRequest.builder().name("test-application-name").build())
+                .logs(ReadRequest.builder().sourceId("test-application-name").build())
                 .as(StepVerifier::create)
                 .expectNext(fill(Log.builder(), "log-message-").build())
                 .expectComplete()
@@ -5254,10 +5257,18 @@ private static void requestListTasksEmpty(
     }
 
     private static void requestLogsRecentLogCache(LogCacheClient logCacheClient, String applicationId) {
-        when(logCacheClient.recentLogs(
-                        ReadRequest.builder().sourceId(applicationId).build()))
-                .thenReturn(
-                        Mono.just(ReadResponse.builder().envelopes(fill(org.cloudfoundry.logcache.v1.EnvelopeBatch.builder()).build()).build()));
+            when(logCacheClient.recentLogs(
+                            ReadRequest.builder().sourceId(applicationId).build()))
+                            .thenReturn(
+                                            Mono.just(fill(ReadResponse.builder())
+                                                     .envelopes(fill(EnvelopeBatch.builder())
+                                                        .batch(fill(org.cloudfoundry.logcache.v1.Envelope.builder())
+                                                            .log(fill(Log.builder())
+                                                                     .payload("test-payload")
+                                                                     .type(LogType.OUT).build())
+                                                            .build())
+                                                        .build())
+                                                     .build()));
     }
 
     private static void requestLogsStream(DopplerClient dopplerClient, String applicationId) {

integration-test/src/test/java/org/cloudfoundry/IntegrationTestConfiguration.java

@@ -46,6 +46,7 @@
 import org.cloudfoundry.client.v2.stacks.ListStacksRequest;
 import org.cloudfoundry.client.v2.userprovidedserviceinstances.CreateUserProvidedServiceInstanceRequest;
 import org.cloudfoundry.doppler.DopplerClient;
+import org.cloudfoundry.logcache.v1.LogCacheClient;
 import org.cloudfoundry.logcache.v1.TestLogCacheEndpoints;
 import org.cloudfoundry.networking.NetworkingClient;
 import org.cloudfoundry.operations.DefaultCloudFoundryOperations;
@@ -254,6 +255,7 @@ ReactorCloudFoundryClient cloudFoundryClient(
     DefaultCloudFoundryOperations cloudFoundryOperations(
             CloudFoundryClient cloudFoundryClient,
             DopplerClient dopplerClient,
+            LogCacheClient logCacheClient,
             NetworkingClient networkingClient,
             RoutingClient routingClient,
             UaaClient uaaClient,
@@ -263,6 +265,7 @@ DefaultCloudFoundryOperations cloudFoundryOperations(
                 .cloudFoundryClient(cloudFoundryClient)
                 .dopplerClient(dopplerClient)
                 .networkingClient(networkingClient)
+                .logCacheClient(logCacheClient)
                 .routingClient(routingClient)
                 .uaaClient(uaaClient)
                 .organization(organizationName)
@@ -547,7 +550,7 @@ Mono<String> stackId(CloudFoundryClient cloudFoundryClient, String stackName) {
 
     @Bean
     String stackName() {
-        return "cflinuxfs3";
+        return "cflinuxfs4";
     }
 
     @Bean(initMethod = "block")

integration-test/src/test/java/org/cloudfoundry/operations/ApplicationsTest.java

@@ -32,6 +32,7 @@
 import org.cloudfoundry.doppler.LogMessage;
 import org.cloudfoundry.doppler.MessageType;
 import org.cloudfoundry.logcache.v1.LogType;
+import org.cloudfoundry.logcache.v1.ReadRequest;
 import org.cloudfoundry.operations.applications.ApplicationDetail;
 import org.cloudfoundry.operations.applications.ApplicationEnvironments;
 import org.cloudfoundry.operations.applications.ApplicationEvent;
@@ -496,17 +497,13 @@ public void logs() throws IOException {
                         this.cloudFoundryOperations,
                         new ClassPathResource("test-application.zip").getFile().toPath(),
                         applicationName,
-                        false)
-                .thenMany(
-                        this.cloudFoundryOperations
+                        false)                
+                .thenMany(this.cloudFoundryOperations
                                 .applications()
-                                .logs(
-                                        LogsRequest.builder()
-                                                .name(applicationName)
-                                                .recent(true)
+                                .logs(ReadRequest.builder()
+                                                .sourceId(applicationName)
                                                 .build()))
                 .map(org.cloudfoundry.logcache.v1.Log::getType)
-                .next()
                 .as(StepVerifier::create)
                 .expectNext(org.cloudfoundry.logcache.v1.LogType.OUT)
                 .expectComplete()