StyxScheduler implements AppInit { @VisibleForTesting static void setupMetrics( PersistentStateManager stateManager, Supplier<Map<WorkflowId, Workflow>> workflowCache, Storage storage, RateLimiter submissionRateLimiter, Stats stats, Time time) { final Supplier<Set<WorkflowId>> enabledWorkflowCache = new CachedSupplier<>(storage::enabled, Instant::now); final CachedSupplier<Map<WorkflowInstance, RunState>> activeStatesCache = new CachedSupplier<>(stateManager::getActiveStates, time); stats.registerWorkflowCountMetric("all", () -> (long) workflowCache.get().size()); stats.registerWorkflowCountMetric("configured", () -> workflowCache.get().values() .stream() .filter(workflow -> workflow.configuration().dockerImage().isPresent()) .count()); stats.registerWorkflowCountMetric("enabled", () -> workflowCache.get().values() .stream() .filter(workflow1 -> workflow1.configuration().dockerImage().isPresent()) .filter(workflow1 -> enabledWorkflowCache.get().contains(WorkflowId.ofWorkflow(workflow1))) .count()); stats.registerWorkflowCountMetric("docker_termination_logging_enabled", () -> workflowCache.get().values() .stream() .filter(workflow -> workflow.configuration().dockerImage().isPresent()) .filter(workflow -> workflow.configuration().dockerTerminationLogging()) .count()); for (State state : State.values()) { for (String triggerType : TriggerUtil.triggerTypesList()) { stats.registerActiveStatesMetric(state, triggerType, () -> activeStatesCache.get().values().stream() .filter(runState -> runState.state().equals(state)) .filter(runState -> runState.data().trigger() .map(TriggerUtil::triggerType) .map(triggerType::equals) .orElse(false)) .count()); } stats.registerActiveStatesMetric(state, "none", () -> activeStatesCache.get().values().stream() .filter(runState -> runState.state().equals(state)) .filter(runState -> runState.data().trigger().isEmpty()) .count()); } stats.registerSubmissionRateLimitMetric(submissionRateLimiter::getRate); } private  StyxScheduler(Builder builder); static Builder newBuilder(); static StyxScheduler createDefault(); @Override void create(Environment environment); @VisibleForTesting static NamespacedKubernetesClient getKubernetesClient(Config config, String id); static final String SERVICE_NAME; static final String STYX_STALE_STATE_TTL_CONFIG; }
@Test public void testSetupMetrics() throws IOException { final WorkflowId wfid1 = WorkflowId.create("foo1", "bar1"); final WorkflowId wfid2 = WorkflowId.create("foo2", "bar2"); final WorkflowConfiguration wfc1 = WorkflowConfiguration.builder() .id(wfid1.id()) .schedule(DAYS) .dockerImage("foo1:bar1") .build(); final WorkflowConfiguration wfc2 = WorkflowConfiguration.builder() .id(wfid2.id()) .schedule(DAYS) .dockerImage("foo2:bar2") .build(); final Workflow wf1 = Workflow.create(wfid1.componentId(), wfc1); final Workflow wf2 = Workflow.create(wfid2.componentId(), wfc2); final WorkflowInstance wfi1 = WorkflowInstance.create(wfid1, "2018-01-02"); final WorkflowInstance wfi2 = WorkflowInstance.create(wfid2, "2018-01-02"); final RunState rs1 = RunState.create(wfi1, State.QUEUED, StateData.newBuilder() .trigger(Trigger.natural()) .build()); final RunState rs2 = RunState.create(wfi2, State.RUNNING, StateData.newBuilder() .trigger(Trigger.adhoc("foobar")) .build()); when(time.get()).thenReturn(Instant.now()); when(stateManager.getActiveStates()).thenReturn(Map.of(wfi1, rs1, wfi2, rs2)); when(workflowCache.get()).thenReturn(Map.of(wfid1, wf1, wfid2, wf2)); when(storage.enabled()).thenReturn(ImmutableSet.of(wfid1)); StyxScheduler.setupMetrics(stateManager, workflowCache, storage, submissionRateLimiter, stats, time); verify(stats).registerWorkflowCountMetric(eq("all"), longGaugeCaptor.capture()); verify(stats).registerWorkflowCountMetric(eq("configured"), longGaugeCaptor.capture()); verify(stats).registerWorkflowCountMetric(eq("enabled"), longGaugeCaptor.capture()); verify(stats).registerWorkflowCountMetric(eq("docker_termination_logging_enabled"), longGaugeCaptor.capture()); for (State state : State.values()) { for (String triggerType : TriggerUtil.triggerTypesList()) { verify(stats).registerActiveStatesMetric(eq(state), eq(triggerType), longGaugeCaptor.capture()); } verify(stats).registerActiveStatesMetric(eq(state), eq("none"), longGaugeCaptor.capture()); } verify(stats).registerSubmissionRateLimitMetric(doubleGaugeCaptor.capture()); longGaugeCaptor.getAllValues().forEach(Gauge::getValue); doubleGaugeCaptor.getAllValues().forEach(Gauge::getValue); verify(stateManager, times(1)).getActiveStates(); verify(storage, times(1)).enabled(); verifyNoMoreInteractions(storage); verifyNoMoreInteractions(stateManager); }