java类com.codahale.metrics.health.HealthCheck的实例源码

/**
 Health check

 @return 204 if healthy otherwise 500
 */
@GET
@Produces(MediaType.APPLICATION_JSON + "; charset=UTF-8")
@Path("check")
public Response check()
{
    for (HealthStatus healthCheck : m_healthCheckService.getChecks())
    {
        HealthCheck.Result result = healthCheck.execute();
        if (!result.isHealthy())
        {
            return setHeaders(Response.status(Response.Status.INTERNAL_SERVER_ERROR)).build();
        }
    }

    return setHeaders(Response.status(m_healthyResponse)).build();
}

public static Handler healthCheckHandler(
    HealthCheckRegistry healthCheckRegistry, ObjectMapper mapper) {
  Preconditions.checkState(healthCheckRegistry != null);
  Preconditions.checkState(mapper != null);

  SortedMap<String, HealthCheck.Result> healthChecks = healthCheckRegistry.runHealthChecks();

  return xrpcRequest ->
      Recipes.newResponseOk(
          xrpcRequest
              .getAlloc()
              .directBuffer()
              .writeBytes(
                  mapper.writerWithDefaultPrettyPrinter().writeValueAsBytes(healthChecks)),
          Recipes.ContentType.Application_Json);
}

public void scheduleHealthChecks(
    EventLoopGroup workerGroup, int initialDelay, int delay, TimeUnit timeUnit) {

  for (Map.Entry<String, HealthCheck> entry : healthCheckMap.entrySet()) {
    healthCheckRegistry.register(entry.getKey(), entry.getValue());
  }

  workerGroup.scheduleWithFixedDelay(
      new Runnable() {
        @Override
        public void run() {
          healthCheckRegistry.runHealthChecks(workerGroup);
        }
      },
      initialDelay,
      delay,
      timeUnit);
}

@Override
public HealthCheck.Result getHealth() {
    try {
        HBaseAdmin.checkHBaseAvailable(configuration);

        return HealthCheck.Result.builder()
                .healthy()
                .withMessage("HBase running on:")
                .withDetail("quorum", quorum)
                .withDetail("clientPort", clientPort)
                .withDetail("znodeParent", znodeParent)
                .build();
    } catch (Exception e) {
        return HealthCheck.Result.builder()
                .unhealthy(e)
                .build();
    }
}

@Override
public HealthCheck.Result getHealth() {
    HealthCheck.ResultBuilder builder = HealthCheck.Result.builder();
    long nonRunningProcessorCount = processors.stream()
            .filter(processor -> !processor.getRunState().equals(RunState.RUNNING))
            .count();
    if (!runState.equals(RunState.RUNNING) || nonRunningProcessorCount > 0) {
        builder.unhealthy();
    } else {
        builder.healthy();
    }
    builder.withDetail("runState", runState.name());
    builder.withDetail("processorCount", processors.size());
    builder.withDetail("processors", processors.stream()
            .collect(HasHealthCheck.buildTreeMapCollector(
                    StatisticsAggregationProcessor::getName,
                    StatisticsAggregationProcessor::produceHealthCheckSummary)));

    return builder.build();
}

@Override
public HealthCheck.Result getHealth() {
    switch (runState) {
        case RUNNING:
            return HealthCheck.Result.builder()
                    .healthy()
                    .withMessage(runState.toString())
                    .withDetail("status", produceHealthCheckSummary())
                    .build();
        default:
            return HealthCheck.Result.builder()
                    .unhealthy()
                    .withMessage(runState.toString())
                    .withDetail("status", produceHealthCheckSummary())
                    .build();
    }
}

@Override
public HealthCheck.Result getHealth() {
    HealthCheck.ResultBuilder builder = HealthCheck.Result.builder();
    long nonRunningProcessorCount = processors.stream()
            .filter(processor -> !processor.getRunState().equals(RunState.RUNNING))
            .count();
    if (!runState.equals(RunState.RUNNING) || nonRunningProcessorCount > 0) {
        builder.unhealthy();
    } else {
        builder.healthy();
    }
    builder.withDetail("runState", runState.name());
    builder.withDetail("processorCount", processors.size());
    builder.withDetail("processors", processors.stream()
            .collect(HasHealthCheck.buildTreeMapCollector(
                    StatisticsFlatMappingProcessor::getName,
                    StatisticsFlatMappingProcessor::produceHealthCheckSummary)));

    return builder.build();
}

@Override
public HealthCheck.Result getHealth() {
    if (stroomPropertyService == null) {
        return HealthCheck.Result.unhealthy("stroomPropertyService has not been initialised");
    } else {

        try {
            //use a treeMap so the props are sorted on output
            return HealthCheck.Result.builder()
                    .withMessage("Available")
                    .withDetail("properties", new TreeMap<>(stroomPropertyService.getAllProperties()))
                    .build();

        } catch (Exception e) {
            return HealthCheck.Result.unhealthy(e);
        }
    }
}

public Map<String, HealthCheck> getHealthChecks() {
    return Collections.singletonMap(
            JOB_MANAGER_JOB_QUEUE_OVERFLOW_HEALTHCHECK,
            new HealthCheck() {
                @Override
                protected Result check() throws Exception {
                    final int queueSize = jobQueue.size();
                    if (queueSize < JOB_MANAGER_MAX_JOB_QUEUE_OVERFLOW_THRESHOLD) {
                        return Result.healthy(format("Queue contains %s entries", queueSize));
                    } else {
                        return Result.unhealthy(format(
                                "%s entries in job queue: this exceeds the warning threshold (%s)",
                                queueSize,
                                JOB_MANAGER_MAX_JOB_QUEUE_OVERFLOW_THRESHOLD));
                    }
                }
            });
}

@Test
public void testGetHealthChecksReturnsAHealthCheckForJobQueueOverflowing() {
    final CancelablePromise<JobExecutionResult> executorPromise = new SimpleCancelablePromise<>();
    final JobManager jobManager = createManagerWith(MockJobExecutor.thatUses(executorPromise));
    final Map<String, HealthCheck> healthChecks = jobManager.getHealthChecks();

    assertThat(healthChecks).containsKeys(JOB_MANAGER_JOB_QUEUE_OVERFLOW_HEALTHCHECK);
    assertThat(healthChecks.get(JOB_MANAGER_JOB_QUEUE_OVERFLOW_HEALTHCHECK)).isNotNull();

    final HealthCheck jobQueueHealthCheck = healthChecks.get(JOB_MANAGER_JOB_QUEUE_OVERFLOW_HEALTHCHECK);

    assertThat(jobQueueHealthCheck.execute().isHealthy());

    for(int i = 0; i < JOB_MANAGER_MAX_JOB_QUEUE_OVERFLOW_THRESHOLD * 2; i++) {
        // These won't finish because we never resolve the promise
        jobManager.submit(STANDARD_VALID_REQUEST);
    }

    assertThat(jobQueueHealthCheck.execute().isHealthy()).isFalse();
}

@Test
public void testZkHealth() throws Exception {

    final CuratorFramework client = newClient(zk.getConnectString(), new RetryOneTime(100));
    client.start();
    client.blockUntilConnected();

    final HealthCheck check = new KafkaHealthCheck(client);
    final HealthCheck.Result res = check.execute();
    assertFalse(res.isHealthy());
    assertTrue(res.getMessage().contains("Error fetching kafka broker list"));

    client.createContainers("/brokers/ids");

    final HealthCheck.Result res2 = check.execute();
    assertFalse(res2.isHealthy());
    assertEquals("No Kafka brokers are connected.", res2.getMessage());

    client.createContainers("/brokers/ids/1");

    final HealthCheck.Result res3 = check.execute();
    assertTrue(res3.isHealthy());
}

@Test
public void testZkHealth() throws Exception {
    final CuratorFramework client = newClient(zk.getConnectString(), new RetryOneTime(100));
    client.start();
    client.blockUntilConnected();

    final HealthCheck check = new ZookeeperHealthCheck(client);
    final HealthCheck.Result res = check.execute();
    assertFalse(res.isHealthy());
    assertEquals("Zookeeper not properly initialized", res.getMessage());

    client.createContainers(ZNODE_COORDINATION);

    final HealthCheck.Result res2 = check.execute();
    assertTrue(res2.isHealthy());
}

/**
 Returns the status of each health check.

 @return 200
 */
@GET
@Produces(MediaType.APPLICATION_JSON + "; charset=UTF-8")
@Path("status")
public Response status()
{
    List<String> messages = new ArrayList<String>();
    for (HealthStatus healthCheck : m_healthCheckService.getChecks())
    {
        HealthCheck.Result result = healthCheck.execute();
        if (result.isHealthy())
        {
            messages.add(healthCheck.getName() + ": OK");
        }
        else
        {
            messages.add(healthCheck.getName() + ": FAIL");
        }
    }

    GenericEntity<List<String>> entity = new GenericEntity<List<String>>(messages)
    {
    };
    return setHeaders(Response.ok(entity)).build();
}

public static void main(String[] args) {
    /*
     *  Init connection pools. They auto register their own health checks.
     */
    ConnectionPools.getProcessing();
    ConnectionPools.getTransactional();

    // Assume some global HttpClient.
    OkHttpClient client = new OkHttpClient.Builder().build();

    HttpUrl passingPath = HttpUrl.parse("http://localhost:8080/ping");
    HealthCheck passing = new ExternalServiceHealthCheck(client, passingPath);
    HealthChecks.getHealthCheckRegistry().register("ping", passing);

    // Since this route doesn't exist it will respond with 404 and should fail the check.
    HttpUrl failingPath = HttpUrl.parse("http://localhost:8080/failingPath");
    HealthCheck failing = new ExternalServiceHealthCheck(client, failingPath);
    HealthChecks.getHealthCheckRegistry().register("shouldFail", failing);

    // Once again pull in a bunch of common middleware.
    SimpleServer server = SimpleServer.simpleServer(Middleware.common(ROUTES));
    server.start();
}

private String healthCheckCommand(IMessage message, OptionSet optionSet) {
    StringBuilder response = new StringBuilder();
    if (healthCheckRegistry.getNames().isEmpty()) {
        return "No health checks registered";
    }
    Map<String, HealthCheck.Result> resultMap = healthCheckRegistry.runHealthChecks();
    response.append("*Health check results*\n");
    for (Map.Entry<String, HealthCheck.Result> entry : resultMap.entrySet()) {
        HealthCheck.Result result = entry.getValue();
        String msg = result.getMessage();
        Throwable t = result.getError();
        response.append(result.isHealthy() ? "[Healthy]" : "[Caution]")
            .append(" **").append(entry.getKey()).append("** ")
            .append(msg != null ? msg : "")
            .append(t != null ? " and exception: *" + t.getMessage() + "*" : "").append("\n");
    }
    return response.toString();
}

private void initDiscordHealthChecks() {
    healthCheckRegistry.register(MetricNames.HEALTH_DISCORD_WS, new HealthCheck() {
        @Override
        protected Result check() throws Exception {
            Counter restartCounter = metricRegistry.counter(MetricNames.DISCORD_WS_RESTARTS);
            Optional<Incident> incident = incidentService.getLastIncidentFromGroup(IncidentService.DISCORD_RESTART);
            ZonedDateTime time = incident.isPresent() ? incident.get().getCreatedDate() : null;
            String reason = incident.isPresent() ? incident.get().getName() : null;
            long restarts = restartCounter.getCount();
            if (restarts > 0) {
                return Result.unhealthy(String.format("%d restart%s, last one on %s (%s)",
                    restarts, restarts == 1 ? "" : "s", time, reason));
            } else {
                return Result.healthy("OK");
            }
        }
    });
}

@GET
@Path(HEALTHCHECK)
@Produces(APPLICATION_JSON)
public Response healthCheck() throws JsonProcessingException {
    SortedMap<String, HealthCheck.Result> results = environment.healthChecks().runHealthChecks();

    Map<String, Map<String, Object>> response = getResponse(results);

    boolean healthy = results.size() == results.values()
            .stream()
            .filter(HealthCheck.Result::isHealthy)
            .count();

    if(healthy) {
        return Response.ok().entity(response).build();
    }
    return status(503).entity(response).build();
}

@Test
public void isHealthyIfNoExceptionIsThrown() throws Exception {
    final Session session = mock(Session.class);
    when(this.factory.openSession()).thenReturn(session);

    final Transaction transaction = mock(Transaction.class);
    when(session.beginTransaction()).thenReturn(transaction);

    final SQLQuery query = mock(SQLQuery.class);
    when(session.createSQLQuery(anyString())).thenReturn(query);

    assertThat(this.healthCheck.execute())
            .isEqualTo(HealthCheck.Result.healthy());

    final InOrder inOrder = inOrder(this.factory, session, transaction, query);
    inOrder.verify(this.factory).openSession();
    inOrder.verify(session).beginTransaction();
    inOrder.verify(session).createSQLQuery("SELECT 1");
    inOrder.verify(query).list();
    inOrder.verify(transaction).commit();
    inOrder.verify(session).close();
}

@Test
public void testHealthCheck() throws Exception {
    ArgumentCaptor<HealthCheck> captor = ArgumentCaptor.forClass(HealthCheck.class);
    verify(_healthChecks, atLeastOnce()).addHealthCheck(Matchers.anyString(), captor.capture());
    List<HealthCheck> healthChecks = captor.getAllValues();

    int numCassandraHealthChecks = 0;
    for (HealthCheck healthCheck : healthChecks) {
        if (healthCheck instanceof CassandraHealthCheck) {
            HealthCheck.Result result = healthCheck.execute();
            assertTrue(result.isHealthy(), result.getMessage());
            numCassandraHealthChecks++;
        }
    }
    assertEquals(numCassandraHealthChecks, 3);  // app, ugc, media
}

@Test
public void testHealthCheck() throws Exception {
    ArgumentCaptor<HealthCheck> captor = ArgumentCaptor.forClass(HealthCheck.class);
    verify(_healthChecks, atLeastOnce()).addHealthCheck(Matchers.anyString(), captor.capture());
    List<HealthCheck> healthChecks = captor.getAllValues();

    int numCassandraHealthChecks = 0;
    for (HealthCheck healthCheck : healthChecks) {
        if (healthCheck instanceof CassandraHealthCheck) {
            HealthCheck.Result result = healthCheck.execute();
            assertTrue(result.isHealthy(), result.getMessage());
            numCassandraHealthChecks++;
        }
    }
    assertEquals(numCassandraHealthChecks, 3);  // app, ugc, databus
}

@Test
public void testHealthCheck() throws Exception {
    ArgumentCaptor<HealthCheck> captor = ArgumentCaptor.forClass(HealthCheck.class);
    verify(_healthChecks, atLeastOnce()).addHealthCheck(Matchers.anyString(), captor.capture());
    List<HealthCheck> healthChecks = captor.getAllValues();

    int numCassandraHealthChecks = 0;
    for (HealthCheck healthCheck : healthChecks) {
        if (healthCheck instanceof CassandraHealthCheck) {
            HealthCheck.Result result = healthCheck.execute();
            assertTrue(result.isHealthy(), result.getMessage());
            numCassandraHealthChecks++;
        }
    }
    assertEquals(numCassandraHealthChecks, 2);  // app, ugc
}

@GET
@Path(HEALTHCHECK)
@Produces(APPLICATION_JSON)
public Response healthCheck() throws JsonProcessingException {
    SortedMap<String, HealthCheck.Result> results = environment.healthChecks().runHealthChecks();

    Map<String, Map<String, Boolean>> response = getResponse(results);

    boolean healthy = results.size() == results.values()
            .stream()
            .filter(HealthCheck.Result::isHealthy)
            .count();

    if(healthy) {
        return Response.ok().entity(response).build();
    }
    return status(503).entity(response).build();
}

@Override
public void run(Environment environment) {
    ServiceLocatorUtilities.bind(serviceLocator, new EnvBinder(application, environment));

    LifecycleEnvironment lifecycle = environment.lifecycle();
    AdminEnvironment admin = environment.admin();

    listServices(HealthCheck.class).forEach(healthCheck -> {
        String name = healthCheck.getClass().getSimpleName();
        environment.healthChecks().register(name, healthCheck);
    });

    listServices(Managed.class).forEach(lifecycle::manage);
    listServices(LifeCycle.class).forEach(lifecycle::manage);
    listServices(LifeCycle.Listener.class).forEach(lifecycle::addLifeCycleListener);
    listServices(ServerLifecycleListener.class).forEach(lifecycle::addServerLifecycleListener);
    listServices(Task.class).forEach(admin::addTask);

    environment.jersey().register(HK2LifecycleListener.class);

    //Set service locator as parent for Jersey's service locator
    environment.getApplicationContext().setAttribute(ServletProperties.SERVICE_LOCATOR, serviceLocator);
    environment.getAdminContext().setAttribute(ServletProperties.SERVICE_LOCATOR, serviceLocator);

    serviceLocator.inject(application);
}

private void handleHealthChecks(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse) throws IOException {
    String healthCheckName = null;
    String path = httpServletRequest.getPathInfo();
    if (StringUtils.isNotBlank(path)) {
        String noLeadingSlash = path.substring(1);
        int indexOfSlash = noLeadingSlash.indexOf("/");
        if (indexOfSlash != -1) {
            healthCheckName = noLeadingSlash.substring(indexOfSlash + 1);
        }
    }

    if (StringUtils.isNotBlank(healthCheckName)) {
        try {
            HealthCheck.Result result = MonitoringCenter.runHealthCheck(healthCheckName);
            writeAsJson(httpServletRequest, httpServletResponse, result);
        } catch (NoSuchElementException e) {
            httpServletResponse.sendError(HttpServletResponse.SC_NOT_FOUND);
            return;
        }
    } else {
        SortedMap<String, HealthCheck.Result> healthCheckResultsByNames = MonitoringCenter.runHealthChecks();
        writeAsJson(httpServletRequest, httpServletResponse, healthCheckResultsByNames);
    }
}

@Test
public void runHealthCheck() throws Exception {
    final boolean isOk = new Date().getTime() % 2 == 0;

    MonitoringCenter.registerHealthCheck("HealthCheckTest1", new HealthCheck() {
        @Override
        protected Result check() throws Exception {
            if (!isOk) {
                return Result.unhealthy("This second is not even!");
            }

            return Result.healthy("Even!");
        }
    });

    HealthCheck.Result healthCheck = MonitoringCenter.runHealthCheck("HealthCheckTest1HealthCheck");
    if (healthCheck.isHealthy()) {
        Assert.assertEquals("Even!", healthCheck.getMessage());
    } else {
        Assert.assertEquals("This second is not even!", healthCheck.getMessage());
    }
    MonitoringCenter.removeAllHealthChecks();
}

@Test
public void testTimerStopsOnceOnException() throws Exception {
    when(metricRegistry.timer(NAME_METHOD)).thenReturn(timer);
    when(metricRegistry.meter(NAME_METHOD + ".errors")).thenReturn(errorMeter);
    when(metricRegistry.counter(NAME_METHOD + ".inFlight")).thenReturn(counter);

    when(timer.time()).thenReturn(context);

    try {
        methodTestStub.faultyMethod();
    } catch (RuntimeException ignored) {}

    verify(healtchCheckRegistry, times(1)).register(eq(NAME_METHOD), any(HealthCheck.class));

    InOrder inOrder = inOrder(context, errorMeter, counter);
    inOrder.verify(counter, times(1)).inc();
    inOrder.verify(context, times(1)).close();
    inOrder.verify(errorMeter, times(1)).mark();
    inOrder.verify(counter, times(1)).dec();

}

@Test
public void testClassAnnotation() throws Exception {
    String methodName = NAME_CLASS + ".faultyMethod";

    when(metricRegistry.timer(methodName)).thenReturn(timer);
    when(metricRegistry.meter(methodName + ".errors")).thenReturn(errorMeter);
    when(metricRegistry.counter(methodName + ".inFlight")).thenReturn(counter);

    when(timer.time()).thenReturn(context);

    try {
        classTestStub.faultyMethod();
    } catch (RuntimeException ignored) {}

    verify(healtchCheckRegistry, times(1)).register(eq(methodName), any(HealthCheck.class));

    InOrder inOrder = inOrder(context, errorMeter, counter);
    inOrder.verify(counter, times(1)).inc();
    inOrder.verify(context, times(1)).close();
    inOrder.verify(errorMeter, times(1)).mark();
    inOrder.verify(counter, times(1)).dec();

}

public static void main(String[] args) {
    /*
     *  Init connection pools. They auto register their own health checks.
     */
    ConnectionPools.getProcessing();
    ConnectionPools.getTransactional();

    // Assume some global HttpClient.
    OkHttpClient client = new OkHttpClient.Builder().build();

    HttpUrl passingPath = HttpUrl.parse("http://localhost:8080/ping");
    HealthCheck passing = new ExternalServiceHealthCheck(client, passingPath);
    HealthChecks.getHealthCheckRegistry().register("ping", passing);

    // Since this route doesn't exist it will respond with 404 and should fail the check.
    HttpUrl failingPath = HttpUrl.parse("http://localhost:8080/failingPath");
    HealthCheck failing = new ExternalServiceHealthCheck(client, failingPath);
    HealthChecks.getHealthCheckRegistry().register("shouldFail", failing);

    // Once again pull in a bunch of common middleware.
    SimpleServer server = SimpleServer.simpleServer(Middleware.common(ROUTES));
    server.start();
}

static HealthCheck.Result unwrap( Result result )
{
    String message = result.getMessage();
    if( result.isHealthy() )
    {
        if( message != null )
        {
            return HealthCheck.Result.healthy( message );
        }
        return HealthCheck.Result.healthy();
    }
    Throwable error = result.getException();
    if( error != null )
    {
        return HealthCheck.Result.unhealthy( error );
    }
    return HealthCheck.Result.unhealthy( message );
}

@GET
@Path("healthcheck")
@Produces(APPLICATION_JSON)
public Response healthCheck() throws JsonProcessingException {
    SortedMap<String, HealthCheck.Result> results = environment.healthChecks().runHealthChecks();

    Map<String, Map<String, Boolean>> response = getResponse(results);

    boolean healthy = results.size() == results.values()
            .stream()
            .filter(HealthCheck.Result::isHealthy)
            .count();

    if (healthy) {
        return Response.ok().entity(response).build();
    }
    return status(503).entity(response).build();
}