@Test public void testCollectionTimeInFuture() throws IOException { BluefloodGauge gauge = new BluefloodGauge("gauge.a.b", 5); long collectionTimeInFuture = new DefaultClockImpl().now().getMillis() + 1000 + Configuration.getInstance().getLongProperty( CoreConfig.AFTER_CURRENT_COLLECTIONTIME_MS ); FullHttpRequest request = createIngestRequest(createRequestBody(TENANT, collectionTimeInFuture, 0 , new BluefloodGauge[]{gauge}, null, null, null)); ArgumentCaptor<FullHttpResponse> argument = ArgumentCaptor.forClass(FullHttpResponse.class); handler.handle(context, request); verify(channel).write(argument.capture()); String errorResponseBody = argument.getValue().content().toString(Charset.defaultCharset()); ErrorResponse errorResponse = getErrorResponse(errorResponseBody); assertEquals("Number of errors invalid", 1, errorResponse.getErrors().size()); assertEquals("Invalid error message", "Out of bounds. Cannot be more than 259200000 milliseconds into the past. " + "Cannot be more than 600000 milliseconds into the future", errorResponse.getErrors().get(0).getMessage()); assertEquals("Invalid source", "timestamp", errorResponse.getErrors().get(0).getSource()); assertEquals("Invalid tenant", TENANT, errorResponse.getErrors().get(0).getTenantId()); assertEquals("Invalid metric name", "", errorResponse.getErrors().get(0).getMetricName()); assertEquals("Invalid status", HttpResponseStatus.BAD_REQUEST, argument.getValue().getStatus()); }
@Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); String body = null; String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); int metricsCount = 0; int delayedMetricsCount = 0; try { body = request.content().toString(Constants.DEFAULT_CHARSET); MetricsCollection collection = new MetricsCollection(); AggregatedPayload payload = AggregatedPayload.create(body); long ingestTime = clock.now().getMillis(); if (payload.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(payload.getTenantId(), payload.getTimestamp(), payload.getDelayTime(ingestTime), payload.getAllMetricNames()); payload.markDelayMetricsReceived(ingestTime); delayedMetricsCount += payload.getAllMetricNames().size(); } else { metricsCount += payload.getAllMetricNames().size(); } List<ErrorResponse.ErrorData> validationErrors = payload.getValidationErrors(); if ( validationErrors.isEmpty() ) { collection.add( PreaggregateConversions.buildMetricsCollection( payload ) ); ListenableFuture<List<Boolean>> futures = processor.apply( collection ); List<Boolean> persisteds = futures.get( timeout.getValue(), timeout.getUnit() ); for ( Boolean persisted : persisteds ) { if (!persisted) { log.error("Internal error persisting data for tenantId:" + payload.getTenantId()); DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); DefaultHandler.sendResponse( ctx, request, null, HttpResponseStatus.OK ); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("JSON request payload: %s", body)); log.error("Error saving data", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } }
HttpAggregatedIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); String body = null; String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); int metricsCount = 0; int delayedMetricsCount = 0; try { body = request.content().toString(Constants.DEFAULT_CHARSET); MetricsCollection collection = new MetricsCollection(); AggregatedPayload payload = AggregatedPayload.create(body); long ingestTime = clock.now().getMillis(); if (payload.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(payload.getTenantId(), payload.getTimestamp(), payload.getDelayTime(ingestTime), payload.getAllMetricNames()); payload.markDelayMetricsReceived(ingestTime); delayedMetricsCount += payload.getAllMetricNames().size(); } else { metricsCount += payload.getAllMetricNames().size(); } List<ErrorResponse.ErrorData> validationErrors = payload.getValidationErrors(); if ( validationErrors.isEmpty() ) { collection.add( PreaggregateConversions.buildMetricsCollection( payload ) ); ListenableFuture<List<Boolean>> futures = processor.apply( collection ); List<Boolean> persisteds = futures.get( timeout.getValue(), timeout.getUnit() ); for ( Boolean persisted : persisteds ) { if (!persisted) { log.error("Internal error persisting data for tenantId:" + payload.getTenantId()); DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); DefaultHandler.sendResponse( ctx, request, null, HttpResponseStatus.OK ); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("JSON request payload: %s", body)); log.error("Error saving data", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } } }
HttpAggregatedIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); String body = null; String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); int metricsCount = 0; int delayedMetricsCount = 0; try { body = request.content().toString(Constants.DEFAULT_CHARSET); MetricsCollection collection = new MetricsCollection(); AggregatedPayload payload = AggregatedPayload.create(body); long ingestTime = clock.now().getMillis(); if (payload.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(payload.getTenantId(), payload.getTimestamp(), payload.getDelayTime(ingestTime), payload.getAllMetricNames()); payload.markDelayMetricsReceived(ingestTime); delayedMetricsCount += payload.getAllMetricNames().size(); } else { metricsCount += payload.getAllMetricNames().size(); } List<ErrorResponse.ErrorData> validationErrors = payload.getValidationErrors(); if ( validationErrors.isEmpty() ) { collection.add( PreaggregateConversions.buildMetricsCollection( payload ) ); ListenableFuture<List<Boolean>> futures = processor.apply( collection ); List<Boolean> persisteds = futures.get( timeout.getValue(), timeout.getUnit() ); for ( Boolean persisted : persisteds ) { if (!persisted) { log.error("Internal error persisting data for tenantId:" + payload.getTenantId()); DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); DefaultHandler.sendResponse( ctx, request, null, HttpResponseStatus.OK ); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("JSON request payload: %s", body)); log.error("Error saving data", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } } HttpAggregatedIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpAggregatedIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); }
HttpAggregatedIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); String body = null; String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); int metricsCount = 0; int delayedMetricsCount = 0; try { body = request.content().toString(Constants.DEFAULT_CHARSET); MetricsCollection collection = new MetricsCollection(); AggregatedPayload payload = AggregatedPayload.create(body); long ingestTime = clock.now().getMillis(); if (payload.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(payload.getTenantId(), payload.getTimestamp(), payload.getDelayTime(ingestTime), payload.getAllMetricNames()); payload.markDelayMetricsReceived(ingestTime); delayedMetricsCount += payload.getAllMetricNames().size(); } else { metricsCount += payload.getAllMetricNames().size(); } List<ErrorResponse.ErrorData> validationErrors = payload.getValidationErrors(); if ( validationErrors.isEmpty() ) { collection.add( PreaggregateConversions.buildMetricsCollection( payload ) ); ListenableFuture<List<Boolean>> futures = processor.apply( collection ); List<Boolean> persisteds = futures.get( timeout.getValue(), timeout.getUnit() ); for ( Boolean persisted : persisteds ) { if (!persisted) { log.error("Internal error persisting data for tenantId:" + payload.getTenantId()); DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); DefaultHandler.sendResponse( ctx, request, null, HttpResponseStatus.OK ); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("JSON request payload: %s", body)); log.error("Error saving data", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } } HttpAggregatedIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpAggregatedIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); @Override void handle(ChannelHandlerContext ctx, FullHttpRequest request); }
HttpAggregatedIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); String body = null; String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); int metricsCount = 0; int delayedMetricsCount = 0; try { body = request.content().toString(Constants.DEFAULT_CHARSET); MetricsCollection collection = new MetricsCollection(); AggregatedPayload payload = AggregatedPayload.create(body); long ingestTime = clock.now().getMillis(); if (payload.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(payload.getTenantId(), payload.getTimestamp(), payload.getDelayTime(ingestTime), payload.getAllMetricNames()); payload.markDelayMetricsReceived(ingestTime); delayedMetricsCount += payload.getAllMetricNames().size(); } else { metricsCount += payload.getAllMetricNames().size(); } List<ErrorResponse.ErrorData> validationErrors = payload.getValidationErrors(); if ( validationErrors.isEmpty() ) { collection.add( PreaggregateConversions.buildMetricsCollection( payload ) ); ListenableFuture<List<Boolean>> futures = processor.apply( collection ); List<Boolean> persisteds = futures.get( timeout.getValue(), timeout.getUnit() ); for ( Boolean persisted : persisteds ) { if (!persisted) { log.error("Internal error persisting data for tenantId:" + payload.getTenantId()); DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); DefaultHandler.sendResponse( ctx, request, null, HttpResponseStatus.OK ); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("JSON request payload: %s", body)); log.error("Error saving data", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } } HttpAggregatedIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpAggregatedIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); @Override void handle(ChannelHandlerContext ctx, FullHttpRequest request);  }