<commit_before>/*
 * Copyright 2013-2015 Rackspace
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

package com.rackspacecloud.blueflood.inputs.handlers;

import com.codahale.metrics.Counter;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.ListenableFuture;
import com.rackspacecloud.blueflood.cache.MetadataCache;
import com.rackspacecloud.blueflood.concurrent.ThreadPoolBuilder;
import com.rackspacecloud.blueflood.http.DefaultHandler;
import com.rackspacecloud.blueflood.http.QueryStringDecoderAndRouter;
import com.rackspacecloud.blueflood.http.RouteMatcher;
import com.rackspacecloud.blueflood.inputs.processors.*;
import com.rackspacecloud.blueflood.io.EventsIO;
import com.rackspacecloud.blueflood.service.*;
import com.rackspacecloud.blueflood.tracker.Tracker;
import com.rackspacecloud.blueflood.types.IMetric;
import com.rackspacecloud.blueflood.types.MetricsCollection;
import com.rackspacecloud.blueflood.utils.Metrics;
import com.rackspacecloud.blueflood.utils.ModuleLoader;
import com.rackspacecloud.blueflood.utils.TimeValue;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class HttpMetricsIngestionServer {
    private static final Logger log = LoggerFactory.getLogger(HttpMetricsIngestionServer.class);
    private static TimeValue DEFAULT_TIMEOUT = new TimeValue(5, TimeUnit.SECONDS);
    private int httpIngestPort;
    private String httpIngestHost;
    private Processor processor;
    private HttpEventsIngestionHandler httpEventsIngestionHandler;
    private final int httpMaxContentLength;

    private TimeValue timeout;

    private EventLoopGroup acceptorGroup;
    private EventLoopGroup workerGroup;
    private ChannelGroup allOpenChannels = new DefaultChannelGroup("allOpenChannels", GlobalEventExecutor.INSTANCE);

    private int HTTP_CONNECTION_READ_IDLE_TIME_SECONDS =
            Configuration.getInstance().getIntegerProperty(HttpConfig.HTTP_CONNECTION_READ_IDLE_TIME_SECONDS);

    public static boolean EXP_TOKEN_SEARCH_IMPROVEMENTS =
            Configuration.getInstance().getBooleanProperty(CoreConfig.ENABLE_TOKEN_SEARCH_IMPROVEMENTS);

    /**
     * Constructor. Instantiate Metrics Ingest server
     * @param context
     */
    public HttpMetricsIngestionServer(ScheduleContext context) {
        this.httpIngestPort = Configuration.getInstance().getIntegerProperty(HttpConfig.HTTP_INGESTION_PORT);
        this.httpIngestHost = Configuration.getInstance().getStringProperty(HttpConfig.HTTP_INGESTION_HOST);
        this.timeout = DEFAULT_TIMEOUT; //TODO: make configurable
        this.processor = new Processor(context, timeout);
        this.httpMaxContentLength = Configuration.getInstance().getIntegerProperty(HttpConfig.HTTP_MAX_CONTENT_LENGTH);

        int acceptThreads = Configuration.getInstance().getIntegerProperty(HttpConfig.MAX_WRITE_ACCEPT_THREADS);
        int workerThreads = Configuration.getInstance().getIntegerProperty(HttpConfig.MAX_WRITE_WORKER_THREADS);
        acceptorGroup = new EpollEventLoopGroup(acceptThreads); // acceptor threads
        workerGroup = new EpollEventLoopGroup(workerThreads);   // client connections threads
    }

    /**
     * Starts the Ingest server
     *
     * @throws InterruptedException
     */
    public void startServer() throws InterruptedException {

        RouteMatcher router = new RouteMatcher();
        router.get("/v1.0", new DefaultHandler());
        router.post("/v1.0/multitenant/experimental/metrics", new HttpMultitenantMetricsIngestionHandler(processor, timeout));
        router.post("/v1.0/:tenantId/experimental/metrics", new HttpMetricsIngestionHandler(processor, timeout));
        router.post("/v1.0/:tenantId/experimental/metrics/statsd", new HttpAggregatedIngestionHandler(processor, timeout));

        router.get("/v2.0", new DefaultHandler());
        router.post("/v2.0/:tenantId/ingest/multi", new HttpMultitenantMetricsIngestionHandler(processor, timeout));
        router.post("/v2.0/:tenantId/ingest", new HttpMetricsIngestionHandler(processor, timeout));
        router.post("/v2.0/:tenantId/ingest/aggregated", new HttpAggregatedIngestionHandler(processor, timeout));
        router.post("/v2.0/:tenantId/events", getHttpEventsIngestionHandler());
        router.post("/v2.0/:tenantId/ingest/aggregated/multi", new HttpAggregatedMultiIngestionHandler(processor, timeout));
        final RouteMatcher finalRouter = router;

        log.info("Starting metrics listener HTTP server on port {}", httpIngestPort);
        ServerBootstrap server = new ServerBootstrap();
        server.group(acceptorGroup, workerGroup)
                .channel(EpollServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        setupPipeline(channel, finalRouter);
                    }
                });

        Channel channel = server.bind(new InetSocketAddress(httpIngestHost, httpIngestPort)).sync().channel();
        allOpenChannels.add(channel);

        //register the tracker MBean for JMX/jolokia
        log.info("Registering tracker service");
        Tracker.getInstance().register();

        log.info("Token search improvements enabled: " + EXP_TOKEN_SEARCH_IMPROVEMENTS);
    }

    private void setupPipeline(SocketChannel channel, RouteMatcher router) {
        final ChannelPipeline pipeline = channel.pipeline();

        pipeline.addLast("logging", new LoggingHandler(LogLevel.TRACE)); //duplex handler
        pipeline.addLast("idleStateHandler", new IdleStateHandler(HTTP_CONNECTION_READ_IDLE_TIME_SECONDS, 0, 0)); //duplex handler
        pipeline.addLast("eventHandler", new UserDefinedEventHandler()); //duplex handler
        pipeline.addLast("encoder", new HttpResponseEncoder()); //outbound handler
        pipeline.addLast("decoder", new HttpRequestDecoder() { //inbound handler

            // if something bad happens during the decode, assume the client send bad data. return a 400.
            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable thr) throws Exception {
                log.debug("Exception during decoding", thr);<commit_msg>undo inadvertent epoll changes<commit_after>/*
 * Copyright 2013-2015 Rackspace
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

package com.rackspacecloud.blueflood.inputs.handlers;

import com.codahale.metrics.Counter;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.ListenableFuture;
import com.rackspacecloud.blueflood.cache.MetadataCache;
import com.rackspacecloud.blueflood.concurrent.ThreadPoolBuilder;
import com.rackspacecloud.blueflood.http.DefaultHandler;
import com.rackspacecloud.blueflood.http.QueryStringDecoderAndRouter;
import com.rackspacecloud.blueflood.http.RouteMatcher;
import com.rackspacecloud.blueflood.inputs.processors.*;
import com.rackspacecloud.blueflood.io.EventsIO;
import com.rackspacecloud.blueflood.service.*;
import com.rackspacecloud.blueflood.tracker.Tracker;
import com.rackspacecloud.blueflood.types.IMetric;
import com.rackspacecloud.blueflood.types.MetricsCollection;
import com.rackspacecloud.blueflood.utils.Metrics;
import com.rackspacecloud.blueflood.utils.ModuleLoader;
import com.rackspacecloud.blueflood.utils.TimeValue;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class HttpMetricsIngestionServer {
    private static final Logger log = LoggerFactory.getLogger(HttpMetricsIngestionServer.class);
    private static TimeValue DEFAULT_TIMEOUT = new TimeValue(5, TimeUnit.SECONDS);
    private int httpIngestPort;
    private String httpIngestHost;
    private Processor processor;
    private HttpEventsIngestionHandler httpEventsIngestionHandler;
    private final int httpMaxContentLength;

    private TimeValue timeout;

    private EventLoopGroup acceptorGroup;
    private EventLoopGroup workerGroup;
    private ChannelGroup allOpenChannels = new DefaultChannelGroup("allOpenChannels", GlobalEventExecutor.INSTANCE);

    private int HTTP_CONNECTION_READ_IDLE_TIME_SECONDS =
            Configuration.getInstance().getIntegerProperty(HttpConfig.HTTP_CONNECTION_READ_IDLE_TIME_SECONDS);

    public static boolean EXP_TOKEN_SEARCH_IMPROVEMENTS =
            Configuration.getInstance().getBooleanProperty(CoreConfig.ENABLE_TOKEN_SEARCH_IMPROVEMENTS);

    /**
     * Constructor. Instantiate Metrics Ingest server
     * @param context
     */
    public HttpMetricsIngestionServer(ScheduleContext context) {
        this.httpIngestPort = Configuration.getInstance().getIntegerProperty(HttpConfig.HTTP_INGESTION_PORT);
        this.httpIngestHost = Configuration.getInstance().getStringProperty(HttpConfig.HTTP_INGESTION_HOST);
        this.timeout = DEFAULT_TIMEOUT; //TODO: make configurable
        this.processor = new Processor(context, timeout);
        this.httpMaxContentLength = Configuration.getInstance().getIntegerProperty(HttpConfig.HTTP_MAX_CONTENT_LENGTH);

        int acceptThreads = Configuration.getInstance().getIntegerProperty(HttpConfig.MAX_WRITE_ACCEPT_THREADS);
        int workerThreads = Configuration.getInstance().getIntegerProperty(HttpConfig.MAX_WRITE_WORKER_THREADS);
        acceptorGroup = new NioEventLoopGroup(acceptThreads); // acceptor threads
        workerGroup = new NioEventLoopGroup(workerThreads);   // client connections threads
    }

    /**
     * Starts the Ingest server
     *
     * @throws InterruptedException
     */
    public void startServer() throws InterruptedException {

        RouteMatcher router = new RouteMatcher();
        router.get("/v1.0", new DefaultHandler());
        router.post("/v1.0/multitenant/experimental/metrics", new HttpMultitenantMetricsIngestionHandler(processor, timeout));
        router.post("/v1.0/:tenantId/experimental/metrics", new HttpMetricsIngestionHandler(processor, timeout));
        router.post("/v1.0/:tenantId/experimental/metrics/statsd", new HttpAggregatedIngestionHandler(processor, timeout));

        router.get("/v2.0", new DefaultHandler());
        router.post("/v2.0/:tenantId/ingest/multi", new HttpMultitenantMetricsIngestionHandler(processor, timeout));
        router.post("/v2.0/:tenantId/ingest", new HttpMetricsIngestionHandler(processor, timeout));
        router.post("/v2.0/:tenantId/ingest/aggregated", new HttpAggregatedIngestionHandler(processor, timeout));
        router.post("/v2.0/:tenantId/events", getHttpEventsIngestionHandler());
        router.post("/v2.0/:tenantId/ingest/aggregated/multi", new HttpAggregatedMultiIngestionHandler(processor, timeout));
        final RouteMatcher finalRouter = router;

        log.info("Starting metrics listener HTTP server on port {}", httpIngestPort);
        ServerBootstrap server = new ServerBootstrap();
        server.group(acceptorGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        setupPipeline(channel, finalRouter);
                    }
                });

        Channel channel = server.bind(new InetSocketAddress(httpIngestHost, httpIngestPort)).sync().channel();
        allOpenChannels.add(channel);

        //register the tracker MBean for JMX/jolokia
        log.info("Registering tracker service");
        Tracker.getInstance().register();

        log.info("Token search improvements enabled: " + EXP_TOKEN_SEARCH_IMPROVEMENTS);
    }

    private void setupPipeline(SocketChannel channel, RouteMatcher router) {
        final ChannelPipeline pipeline = channel.pipeline();

        pipeline.addLast("logging", new LoggingHandler(LogLevel.TRACE)); //duplex handler
        pipeline.addLast("idleStateHandler", new IdleStateHandler(HTTP_CONNECTION_READ_IDLE_TIME_SECONDS, 0, 0)); //duplex handler
        pipeline.addLast("eventHandler", new UserDefinedEventHandler()); //duplex handler
        pipeline.addLast("encoder", new HttpResponseEncoder()); //outbound handler
        pipeline.addLast("decoder", new HttpRequestDecoder() { //inbound handler

            // if something bad happens during the decode, assume the client send bad data. return a 400.
            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable thr) throws Exception {
                log.debug("Exception during decoding", thr);
{'fraction_non_alphanumeric': 0.0910057127673708, 'fraction_numerical': 0.005314202205393915, 'mean_word_length': 4.216562716562716, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2898858', 'n_tokens_mistral': 4112, 'n_tokens_neox': 4015, 'n_words': 842}