blueflood-http/src/main/java/com/rackspacecloud/blueflood/inputs/handlers/HttpAggregatedMultiIngestionHandler.java
@Override
    public void handle(ChannelHandlerContext ctx, FullHttpRequest request) {

        Tracker.getInstance().track(request);

        requestCount.inc();

        final Timer.Context timerContext = handlerTimer.time();
        long ingestTime = clock.now().getMillis();

        // this is all JSON.
        String body = null;
        try {
            String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER);

            body = request.content().toString(Constants.DEFAULT_CHARSET);
            List<AggregatedPayload> bundleList = createBundleList(body);

            if (bundleList.size() > 0) {
                // has aggregated metric bundle in body
                // convert and add metric bundle to MetricsCollection if valid
                MetricsCollection collection = new MetricsCollection();
                List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>();

                // for each metric bundle
                int delayedMetricsCount = 0;
                int metricsCount = 0;
                for (AggregatedPayload bundle : bundleList) {
                    // validate, convert, and add to collection
                    List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors();
                    if (bundleValidationErrors.isEmpty()) {
                        // no validation error, add to collection
                        collection.add(PreaggregateConversions.buildMetricsCollection(bundle));
                    } else {
                        // failed validation, add to error
                        errors.addAll(bundleValidationErrors);
                    }

                    if (bundle.hasDelayedMetrics(ingestTime)) {
                        Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(),
                                bundle.getTimestamp(),
                                bundle.getDelayTime(ingestTime),
                                bundle.getAllMetricNames());
                        bundle.markDelayMetricsReceived(ingestTime);
                        delayedMetricsCount += bundle.getAllMetricNames().size();
                    } else {
                        metricsCount += bundle.getAllMetricNames().size();
                    }
                }

                // if has validation errors and no valid metrics
                if (!errors.isEmpty() && collection.size() == 0) {
                    // return BAD_REQUEST and error
                    DefaultHandler.sendErrorResponse(ctx, request,
                            errors,
                            HttpResponseStatus.BAD_REQUEST);
                    return;
                }

                // process valid metrics in collection
                ListenableFuture<List<Boolean>> futures = processor.apply(collection);
                List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit());
                for (Boolean persisted : persisteds) {
                    if (!persisted) {
                        DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR);
                        return;
                    }
                }

                recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount);

                // return OK or MULTI_STATUS response depending if there were validation errors
                if (errors.isEmpty()) {
                    // no validation error, response OK
                    DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK);
                    return;
                } else {
                    // has some validation errors, response MULTI_STATUS
                    DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS);
                    return;
                }

            } else {
                // no aggregated metric bundles in body, response OK
                DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST);
                return;
            }
        } 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("Exception processing: %s", body));
            log.error("Other exception while trying to parse content", ex);
            DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR);
        } finally {
            timerContext.stop();
            requestCount.dec();
        }

    }
@Override
    public void handle(ChannelHandlerContext ctx, FullHttpRequest request) {

        Tracker.getInstance().track(request);

        requestCount.inc();

        final Timer.Context timerContext = handlerTimer.time();
        long ingestTime = clock.now().getMillis();

        // this is all JSON.
        String body = null;
        try {
            String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER);

            body = request.content().toString(Constants.DEFAULT_CHARSET);
            List<AggregatedPayload> bundleList = createBundleList(body);

            if (bundleList.size() > 0) {
                // has aggregated metric bundle in body
                // convert and add metric bundle to MetricsCollection if valid
                MetricsCollection collection = new MetricsCollection();
                List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>();

                // for each metric bundle
                int delayedMetricsCount = 0;
                int metricsCount = 0;
                for (AggregatedPayload bundle : bundleList) {
                    // validate, convert, and add to collection
                    List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors();
                    if (bundleValidationErrors.isEmpty()) {
                        // no validation error, add to collection
                        collection.add(PreaggregateConversions.buildMetricsCollection(bundle));
                    } else {
                        // failed validation, add to error
                        errors.addAll(bundleValidationErrors);
                    }

                    if (bundle.hasDelayedMetrics(ingestTime)) {
                        Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(),
                                bundle.getTimestamp(),
                                bundle.getDelayTime(ingestTime),
                                bundle.getAllMetricNames());
                        bundle.markDelayMetricsReceived(ingestTime);
                        delayedMetricsCount += bundle.getAllMetricNames().size();
                    } else {
                        metricsCount += bundle.getAllMetricNames().size();
                    }
                }

                // if has validation errors and no valid metrics
                if (!errors.isEmpty() && collection.size() == 0) {
                    // return BAD_REQUEST and error
                    DefaultHandler.sendErrorResponse(ctx, request,
                            errors,
                            HttpResponseStatus.BAD_REQUEST);
                    return;
                }

                // process valid metrics in collection
                ListenableFuture<List<Boolean>> futures = processor.apply(collection);
                List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit());
                for (Boolean persisted : persisteds) {
                    if (!persisted) {
                        DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR);
                        return;
                    }
                }

                recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount);

                // return OK or MULTI_STATUS response depending if there were validation errors
                if (errors.isEmpty()) {
                    // no validation error, response OK
                    DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK);
                    return;
                } else {
                    // has some validation errors, response MULTI_STATUS
                    DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS);
                    return;
                }

            } else {
                // no aggregated metric bundles in body, response OK
                DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST);
                return;
            }
        } 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("Exception processing: %s", body));
            log.error("Other exception while trying to parse content", ex);
            DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR);
        } finally {
            timerContext.stop();
            requestCount.dec();
        }

    }
@ Override public void handle ( ChannelHandlerContext ctx , FullHttpRequest request ) { Tracker . getInstance ( ) . track ( request ) ; requestCount . inc ( ) ; final Timer . Context timerContext = handlerTimer . time ( ) ; long ingestTime = clock . now ( ) . getMillis ( ) ; // this is all JSON. String body = null ; try { String submitterTenantId = request . headers ( ) . get ( HttpMetricsIngestionServer . TENANT_ID_HEADER ) ; body = request . content ( ) . toString ( Constants . DEFAULT_CHARSET ) ; List < AggregatedPayload > bundleList = createBundleList ( body ) ; if ( bundleList . size ( ) > 0 ) { // has aggregated metric bundle in body // convert and add metric bundle to MetricsCollection if valid MetricsCollection collection = new MetricsCollection ( ) ; List < ErrorResponse . ErrorData > errors = new ArrayList < ErrorResponse . ErrorData > ( ) ; // for each metric bundle int delayedMetricsCount = 0 ; int metricsCount = 0 ; for ( AggregatedPayload bundle : bundleList ) { // validate, convert, and add to collection List < ErrorResponse . ErrorData > bundleValidationErrors = bundle . getValidationErrors ( ) ; if ( bundleValidationErrors . isEmpty ( ) ) { // no validation error, add to collection collection . add ( PreaggregateConversions . buildMetricsCollection ( bundle ) ) ; } else { // failed validation, add to error errors . addAll ( bundleValidationErrors ) ; } if ( bundle . hasDelayedMetrics ( ingestTime ) ) { Tracker . getInstance ( ) . trackDelayedAggregatedMetricsTenant ( bundle . getTenantId ( ) , bundle . getTimestamp ( ) , bundle . getDelayTime ( ingestTime ) , bundle . getAllMetricNames ( ) ) ; bundle . markDelayMetricsReceived ( ingestTime ) ; delayedMetricsCount += bundle . getAllMetricNames ( ) . size ( ) ; } else { metricsCount += bundle . getAllMetricNames ( ) . size ( ) ; } } // if has validation errors and no valid metrics if ( ! errors . isEmpty ( ) && collection . size ( ) == 0 ) { // return BAD_REQUEST and error DefaultHandler . sendErrorResponse ( ctx , request , errors , HttpResponseStatus . BAD_REQUEST ) ; return ; } // process valid metrics in collection ListenableFuture < List < Boolean > > futures = processor . apply ( collection ) ; List < Boolean > persisteds = futures . get ( timeout . getValue ( ) , timeout . getUnit ( ) ) ; for ( Boolean persisted : persisteds ) { if ( ! persisted ) { DefaultHandler . sendErrorResponse ( ctx , request , "Internal error persisting data" , HttpResponseStatus . INTERNAL_SERVER_ERROR ) ; return ; } } recordPerTenantMetrics ( submitterTenantId , metricsCount , delayedMetricsCount ) ; // return OK or MULTI_STATUS response depending if there were validation errors if ( errors . isEmpty ( ) ) { // no validation error, response OK DefaultHandler . sendResponse ( ctx , request , null , HttpResponseStatus . OK ) ; return ; } else { // has some validation errors, response MULTI_STATUS DefaultHandler . sendErrorResponse ( ctx , request , errors , HttpResponseStatus . MULTI_STATUS ) ; return ; } } else { // no aggregated metric bundles in body, response OK DefaultHandler . sendResponse ( ctx , request , "No valid metrics" , HttpResponseStatus . BAD_REQUEST ) ; return ; } } 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 ( "Exception processing: %s" , body ) ) ; log . error ( "Other exception while trying to parse content" , ex ) ; DefaultHandler . sendErrorResponse ( ctx , request , "Internal error saving data" , HttpResponseStatus . INTERNAL_SERVER_ERROR ) ; } finally { timerContext . stop ( ) ; requestCount . dec ( ) ; } }
our own stuff.
our own stuff .