DeployBroker extends BrokerPhase { @Override public List<ChePlugin> execute() throws InfrastructureException { LOG.debug("Starting brokers pod for workspace '{}'", runtimeId.getWorkspaceId()); Span tracingSpan = tracer.buildSpan(DEPLOY_BROKER_PHASE).start(); TracingTags.WORKSPACE_ID.set(tracingSpan, runtimeId.getWorkspaceId()); KubernetesDeployments deployments = namespace.deployments(); try { for (ConfigMap configMap : brokerEnvironment.getConfigMaps().values()) { namespace.configMaps().create(configMap); } for (Secret secret : brokerEnvironment.getSecrets().values()) { namespace.secrets().create(secret); } Pod pluginBrokerPod = getPluginBrokerPod(brokerEnvironment.getPodsCopy()); if (factory.isConfigured()) { UnrecoverablePodEventListener unrecoverableEventListener = factory.create( ImmutableSet.of(pluginBrokerPod.getMetadata().getName()), this::handleUnrecoverableEvent); namespace.deployments().watchEvents(unrecoverableEventListener); } namespace .deployments() .watchEvents( new RuntimeLogsPublisher( runtimeEventsPublisher, runtimeId, ImmutableSet.of(pluginBrokerPod.getMetadata().getName()))); deployments.create(pluginBrokerPod); watchLogsIfDebugEnabled(startOptions, pluginBrokerPod); LOG.debug("Brokers pod is created for workspace '{}'", runtimeId.getWorkspaceId()); tracingSpan.finish(); return nextPhase.execute(); } catch (InfrastructureException e) { namespace.deployments().stopWatch(true); TracingTags.setErrorStatus(tracingSpan, e); tracingSpan.finish(); throw e; } finally { namespace.deployments().stopWatch(); try { deployments.delete(); } catch (InfrastructureException e) { LOG.error("Brokers pod removal failed. Error: " + e.getLocalizedMessage(), e); } try { namespace.secrets().delete(); } catch (InfrastructureException ex) { LOG.error("Brokers secret removal failed. Error: " + ex.getLocalizedMessage(), ex); } try { namespace.configMaps().delete(); } catch (InfrastructureException ex) { LOG.error("Brokers config map removal failed. Error: " + ex.getLocalizedMessage(), ex); } } } DeployBroker(
      RuntimeIdentity runtimeId,
      KubernetesNamespace namespace,
      KubernetesEnvironment brokerEnvironment,
      BrokersResult brokersResult,
      UnrecoverablePodEventListenerFactory factory,
      RuntimeEventsPublisher runtimeEventsPublisher,
      Tracer tracer,
      Map<String, String> startOptions); @Override List<ChePlugin> execute();  }
@Test public void shouldDeployPluginBrokerEnvironment() throws Exception { List<ChePlugin> result = deployBrokerPhase.execute(); assertSame(result, plugins); verify(k8sConfigMaps).create(configMap); verify(k8sDeployments).create(pod); verify(k8sSecrets).create(tlsSecret); verify(k8sDeployments).stopWatch(); verify(k8sDeployments).delete(); verify(k8sConfigMaps).delete(); verify(k8sSecrets).delete(); }
@Test public void shouldListenToUnrecoverableEventsIfFactoryIsConfigured() throws Exception { when(unrecoverableEventListenerFactory.isConfigured()).thenReturn(true); UnrecoverablePodEventListener listener = mock(UnrecoverablePodEventListener.class); when(unrecoverableEventListenerFactory.create(any(Set.class), any())).thenReturn(listener); deployBrokerPhase.execute(); verify(unrecoverableEventListenerFactory).isConfigured(); verify(unrecoverableEventListenerFactory) .create(eq(ImmutableSet.of(PLUGIN_BROKER_POD_NAME)), any()); verify(k8sDeployments).watchEvents(listener); verify(k8sDeployments).stopWatch(); }
@Test public void shouldListenToPodEventsToPropagateThemAsLogs() throws Exception { deployBrokerPhase.execute(); verify(k8sDeployments).watchEvents(any(RuntimeLogsPublisher.class)); verify(k8sDeployments).stopWatch(); }
@Test public void shouldWatchTheLogsWhenSetInStartOptions() throws Exception { deployBrokerPhase = new DeployBroker( RUNTIME_ID, k8sNamespace, k8sEnvironment, brokersResult, unrecoverableEventListenerFactory, runtimeEventPublisher, tracer, singletonMap(DEBUG_WORKSPACE_START, Boolean.TRUE.toString())); deployBrokerPhase.then(nextBrokerPhase); when(nextBrokerPhase.execute()).thenReturn(plugins); deployBrokerPhase.execute(); verify(k8sDeployments) .watchLogs( any(PodLogHandler.class), any(RuntimeEventsPublisher.class), any(LogWatchTimeouts.class), any(), eq(LogWatcher.DEFAULT_LOG_LIMIT_BYTES)); verify(k8sDeployments).stopWatch(); }
@Test public void shouldWatchTheLogsWithProperBytesLimitWhenSet() throws Exception { deployBrokerPhase = new DeployBroker( RUNTIME_ID, k8sNamespace, k8sEnvironment, brokersResult, unrecoverableEventListenerFactory, runtimeEventPublisher, tracer, ImmutableMap.of( DEBUG_WORKSPACE_START, Boolean.TRUE.toString(), Constants.DEBUG_WORKSPACE_START_LOG_LIMIT_BYTES, "123")); deployBrokerPhase.then(nextBrokerPhase); when(nextBrokerPhase.execute()).thenReturn(plugins); deployBrokerPhase.execute(); verify(k8sDeployments) .watchLogs( any(PodLogHandler.class), any(RuntimeEventsPublisher.class), any(LogWatchTimeouts.class), any(), eq(123L)); verify(k8sDeployments).stopWatch(); }
@Test public void shouldNotWatchTheLogsWhenSetFalseInStartOptions() throws Exception { deployBrokerPhase = new DeployBroker( RUNTIME_ID, k8sNamespace, k8sEnvironment, brokersResult, unrecoverableEventListenerFactory, runtimeEventPublisher, tracer, singletonMap(DEBUG_WORKSPACE_START, Boolean.FALSE.toString())); deployBrokerPhase.then(nextBrokerPhase); when(nextBrokerPhase.execute()).thenReturn(plugins); deployBrokerPhase.execute(); verify(k8sDeployments, never()) .watchLogs( any(PodLogHandler.class), any(RuntimeEventsPublisher.class), any(LogWatchTimeouts.class), any(), eq(LogWatcher.DEFAULT_LOG_LIMIT_BYTES)); verify(k8sDeployments).stopWatch(); }
@Test public void shouldNotWatchTheLogsWhenEmptyStartOptions() throws Exception { deployBrokerPhase.execute(); verify(k8sDeployments, never()) .watchLogs( any(PodLogHandler.class), any(RuntimeEventsPublisher.class), any(LogWatchTimeouts.class), any(), eq(LogWatcher.DEFAULT_LOG_LIMIT_BYTES)); verify(k8sDeployments).stopWatch(); }
@Test public void shouldNotWatchTheLogsWhenNullStartOptions() throws Exception { deployBrokerPhase = new DeployBroker( RUNTIME_ID, k8sNamespace, k8sEnvironment, brokersResult, unrecoverableEventListenerFactory, runtimeEventPublisher, tracer, null); deployBrokerPhase.then(nextBrokerPhase); when(nextBrokerPhase.execute()).thenReturn(plugins); deployBrokerPhase.execute(); verify(k8sDeployments, never()) .watchLogs( any(PodLogHandler.class), any(RuntimeEventsPublisher.class), any(LogWatchTimeouts.class), any(), eq(LogWatcher.DEFAULT_LOG_LIMIT_BYTES)); verify(k8sDeployments).stopWatch(); }
@Test public void shouldDoNotListenToUnrecoverableEventsIfFactoryIsConfigured() throws Exception { when(unrecoverableEventListenerFactory.isConfigured()).thenReturn(false); deployBrokerPhase.execute(); verify(unrecoverableEventListenerFactory).isConfigured(); verify(unrecoverableEventListenerFactory, never()) .create(eq(ImmutableSet.of(PLUGIN_BROKER_POD_NAME)), any()); verify(k8sDeployments, never()).watchEvents(any(UnrecoverablePodEventListener.class)); verify(k8sDeployments).stopWatch(); }
@Test( expectedExceptions = InternalInfrastructureException.class, expectedExceptionsMessageRegExp = "Plugin broker environment must have only one pod\\. Workspace `workspaceId` contains `0` pods\\.") public void shouldThrowExceptionIfThereIsNoAnyPodsInEnvironment() throws Exception { when(k8sEnvironment.getPodsCopy()).thenReturn(emptyMap()); deployBrokerPhase.execute(); }
@Test( expectedExceptions = InternalInfrastructureException.class, expectedExceptionsMessageRegExp = "Plugin broker environment must have only one pod\\. Workspace `workspaceId` contains `2` pods\\.") public void shouldThrowExceptionIfThereAreMoreThanOnePodsInEnvironment() throws Exception { when(k8sEnvironment.getPodsCopy()).thenReturn(ImmutableMap.of("pod1", pod, "pod2", pod)); deployBrokerPhase.execute(); }