ContainerLogWatch implements Runnable, Closeable { @Override public void close() { synchronized (this) { closed = true; if (currentLogWatch != null) { currentLogWatch.close(); } } } ContainerLogWatch(
      KubernetesClient client,
      RuntimeEventsPublisher eventPublisher,
      String namespace,
      String podName,
      String containerName,
      PodLogHandler logHandler,
      LogWatchTimeouts timeouts,
      long inputStreamLimit); @Override void run(); @Override void close(); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString();  }
@Test public void testCloseFromOutside() throws IOException, InterruptedException { PipedInputStream inputStream = new PipedInputStream(); PipedOutputStream outputStream = new PipedOutputStream(inputStream); outputStream.write("message\n".getBytes()); logWatch.setInputStream(inputStream); CountDownLatch latch = new CountDownLatch(1); doAnswer( (a) -> { outputStream.write("nextMessage\n".getBytes()); latch.countDown(); return null; }) .when(podLogHandler) .handle(any(), any()); ContainerLogWatch clw = new ContainerLogWatch( client, eventsPublisher, namespace, podname, container, podLogHandler, TIMEOUTS, LOG_LIMIT_BYTES); new Thread(clw).start(); latch.await(1, TimeUnit.SECONDS); clw.close(); assertTrue(logWatch.isClosed); verify(eventsPublisher, times(1)).sendWatchLogStartedEvent(any(String.class)); verify(eventsPublisher, timeout(1000).times(1)).sendWatchLogStoppedEvent(any(String.class)); }
@Test public void testCloseOfOutputStream() throws IOException, InterruptedException { PipedInputStream inputStream = new PipedInputStream(); PipedOutputStream outputStream = new PipedOutputStream(inputStream); outputStream.write("message\n".getBytes()); logWatch.setInputStream(inputStream); CountDownLatch messageHandleLatch = new CountDownLatch(1); doAnswer( (a) -> { messageHandleLatch.countDown(); return null; }) .when(podLogHandler) .handle("message", container); ContainerLogWatch clw = new ContainerLogWatch( client, eventsPublisher, namespace, podname, container, podLogHandler, TIMEOUTS, LOG_LIMIT_BYTES); new Thread(clw).start(); messageHandleLatch.await(1, TimeUnit.SECONDS); outputStream.close(); CountDownLatch closeLatch = new CountDownLatch(1); logWatch.setLatch(closeLatch); closeLatch.await(1, TimeUnit.SECONDS); assertTrue(logWatch.isClosed); verify(eventsPublisher, times(1)).sendWatchLogStartedEvent(any(String.class)); verify(eventsPublisher, timeout(1000).times(1)).sendWatchLogStoppedEvent(any(String.class)); }
@Test public void shouldRetryWhenErrorMessageReceived() throws IOException, InterruptedException { String podInitializingMessage = "{\"kind\":\"Status\"," + "\"apiVersion\":\"v1\"," + "\"metadata\":{}," + "\"status\":\"Failure\"," + "\"message\":\"container \\\"" + container + "\\\" in pod \\\"" + podname + "\\\" is waiting to start: ContainerCreating\"," + "\"reason\":\"BadRequest\"," + "\"code\":400}"; PipedInputStream inputStream = new PipedInputStream(); PipedOutputStream outputStream = new PipedOutputStream(inputStream); outputStream.write((podInitializingMessage + "\n").getBytes()); outputStream.close(); logWatch.setInputStream(inputStream); LogWatchMock logWatchRegularMessage = new LogWatchMock(); inputStream = new PipedInputStream(); outputStream = new PipedOutputStream(inputStream); outputStream.write("message\n".getBytes()); outputStream.close(); logWatchRegularMessage.setInputStream(inputStream); CountDownLatch messageHandleLatch = new CountDownLatch(2); logWatchRegularMessage.setLatch(messageHandleLatch); doAnswer( (a) -> { messageHandleLatch.countDown(); return null; }) .when(podLogHandler) .handle("message", container); when(pods.watchLog()).thenReturn(logWatch).thenReturn(logWatchRegularMessage); ContainerLogWatch clw = new ContainerLogWatch( client, eventsPublisher, namespace, podname, container, podLogHandler, TIMEOUTS, LOG_LIMIT_BYTES); new Thread(clw).start(); CountDownLatch closeLatch = new CountDownLatch(1); logWatch.setLatch(closeLatch); closeLatch.await(1, TimeUnit.SECONDS); assertTrue(logWatch.isClosed); messageHandleLatch.await(1, TimeUnit.SECONDS); verify(podLogHandler).handle("message", container); assertTrue(logWatchRegularMessage.isClosed); verify(eventsPublisher, times(2)).sendWatchLogStartedEvent(any(String.class)); verify(eventsPublisher, timeout(1000).times(2)).sendWatchLogStoppedEvent(any(String.class)); }
@Test public void shouldRetryWhenOutputIsNullFirst() throws IOException, InterruptedException { logWatch.setInputStream(null); LogWatchMock logWatchRegularMessage = new LogWatchMock(); PipedInputStream inputStream = new PipedInputStream(); PipedOutputStream outputStream = new PipedOutputStream(inputStream); outputStream.write("message\n".getBytes()); outputStream.close(); logWatchRegularMessage.setInputStream(inputStream); CountDownLatch messageHandleLatch = new CountDownLatch(2); logWatchRegularMessage.setLatch(messageHandleLatch); doAnswer( (a) -> { messageHandleLatch.countDown(); return null; }) .when(podLogHandler) .handle("message", container); when(pods.watchLog()).thenReturn(logWatch).thenReturn(logWatchRegularMessage); ContainerLogWatch clw = new ContainerLogWatch( client, eventsPublisher, namespace, podname, container, podLogHandler, TIMEOUTS, LOG_LIMIT_BYTES); new Thread(clw).start(); CountDownLatch closeLatch = new CountDownLatch(1); logWatch.setLatch(closeLatch); closeLatch.await(1, TimeUnit.SECONDS); assertTrue(logWatch.isClosed); messageHandleLatch.await(1, TimeUnit.SECONDS); verify(podLogHandler).handle("message", container); assertTrue(logWatchRegularMessage.isClosed); verify(eventsPublisher, times(2)).sendWatchLogStartedEvent(any(String.class)); verify(eventsPublisher, timeout(1000).times(2)).sendWatchLogStoppedEvent(any(String.class)); }