<commit_before>/*
 * Copyright (c) 2016 Ericsson India Global Services Pvt Ltd. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */
package org.opendaylight.netvirt.natservice.internal;

import com.google.common.primitives.Ints;
import java.math.BigInteger;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import javax.inject.Singleton;
import org.opendaylight.controller.liblldp.NetUtils;
import org.opendaylight.genius.mdsalutil.MetaDataUtil;
import org.opendaylight.genius.mdsalutil.NWUtil;
import org.opendaylight.genius.mdsalutil.NwConstants;
import org.opendaylight.genius.mdsalutil.packet.Ethernet;
import org.opendaylight.genius.mdsalutil.packet.IPv4;
import org.opendaylight.genius.mdsalutil.packet.TCP;
import org.opendaylight.genius.mdsalutil.packet.UDP;
import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketProcessingListener;
import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Singleton
public class NaptPacketInHandler implements PacketProcessingListener {

    private static final Logger LOG = LoggerFactory.getLogger(NaptPacketInHandler.class);
    private final ConcurrentMap<String,NatPacketProcessingState> incomingPacketMap = new ConcurrentHashMap<>();
    private final NaptEventHandler naptEventHandler;
    private ExecutorService firstPacketExecutorService;
    private ExecutorService retryPacketExecutorService;

    @Inject
    public NaptPacketInHandler(NaptEventHandler naptEventHandler) {
        this.naptEventHandler = naptEventHandler;
    }

    @PostConstruct
    public void init() {
        firstPacketExecutorService = Executors.newFixedThreadPool(NatConstants.SNAT_PACKET_THEADPOOL_SIZE);
        retryPacketExecutorService = Executors.newFixedThreadPool(NatConstants.SNAT_PACKET_RETRY_THEADPOOL_SIZE);
    }

    @PreDestroy
    public void close() {
        if (firstPacketExecutorService != null) {
            firstPacketExecutorService.shutdown();
        }

        if (retryPacketExecutorService != null) {
            retryPacketExecutorService.shutdown();
        }
    }

    @Override
    // TODO Clean up the exception handling
    @SuppressWarnings("checkstyle:IllegalCatch")
    public void onPacketReceived(PacketReceived packetReceived) {
        String internalIPAddress = "";
        int portNumber = 0;
        long routerId = 0L;
        NAPTEntryEvent.Operation operation = NAPTEntryEvent.Operation.ADD;
        NAPTEntryEvent.Protocol protocol;

        Short tableId = packetReceived.getTableId().getValue();

        LOG.trace("onPacketReceived : packet: {}, tableId {}", packetReceived, tableId);

        if (tableId == NwConstants.OUTBOUND_NAPT_TABLE) {
            LOG.debug("onPacketReceived : NAPTPacketInHandler Packet for Outbound NAPT Table");
            byte[] inPayload = packetReceived.getPayload();
            Ethernet ethPkt = new Ethernet();
            if (inPayload != null) {
                try {
                    ethPkt.deserialize(inPayload, 0, inPayload.length * NetUtils.NumBitsInAByte);
                } catch (Exception e) {
                    LOG.warn("onPacketReceived: Failed to decode Packet", e);
                    return;
                }
                if (ethPkt.getPayload() instanceof IPv4) {
                    IPv4 ipPkt = (IPv4) ethPkt.getPayload();
                    byte[] ipSrc = Ints.toByteArray(ipPkt.getSourceAddress());

                    internalIPAddress = NWUtil.toStringIpAddress(ipSrc);
                    LOG.trace("onPacketReceived : Retrieved internalIPAddress {}", internalIPAddress);
                    if (ipPkt.getPayload() instanceof TCP) {
                        TCP tcpPkt = (TCP) ipPkt.getPayload();
                        portNumber = tcpPkt.getSourcePort();
                        if (portNumber < 0) {
                            portNumber = 32767 + portNumber + 32767 + 2;
                            LOG.trace("onPacketReceived : Retrieved and extracted TCP portNumber {}", portNumber);
                        }
                        protocol = NAPTEntryEvent.Protocol.TCP;
                        LOG.trace("onPacketReceived : Retrieved TCP portNumber {}", portNumber);
                    } else if (ipPkt.getPayload() instanceof UDP) {
                        UDP udpPkt = (UDP) ipPkt.getPayload();
                        portNumber = udpPkt.getSourcePort();
                        if (portNumber < 0) {
                            portNumber = 32767 + portNumber + 32767 + 2;
                            LOG.trace("onPacketReceived : Retrieved and extracted UDP portNumber {}", portNumber);
                        }
                        protocol = NAPTEntryEvent.Protocol.UDP;
                        LOG.trace("onPacketReceived : Retrieved UDP portNumber {}", portNumber);
                    } else {
                        LOG.error("onPacketReceived : Incoming Packet is neither TCP or UDP packet");
                        return;
                    }
                } else {
                    LOG.error("onPacketReceived : Incoming Packet is not IPv4 packet");
                    return;
                }

                if (internalIPAddress != null) {
                    BigInteger metadata = packetReceived.getMatch().getMetadata().getMetadata();
                    routerId = MetaDataUtil.getNatRouterIdFromMetadata(metadata);
                    if (routerId <= 0) {
                        LOG.error("onPacketReceived : Router ID is invalid");
                        return;
                    }
                    String sourceIPPortKey = routerId + NatConstants.COLON_SEPARATOR
                            + internalIPAddress + NatConstants.COLON_SEPARATOR + portNumber;

                    NatPacketProcessingState state = incomingPacketMap.get(sourceIPPortKey);
                    if (state == null) {
                        state = new NatPacketProcessingState(System.currentTimeMillis(), -1);
                        incomingPacketMap.put(sourceIPPortKey, state);
                        LOG.trace("onPacketReceived : Processing new SNAT({}) Packet", sourceIPPortKey);

                        //send to Event Queue
                        NAPTEntryEvent naptEntryEvent = new NAPTEntryEvent(internalIPAddress, portNumber, routerId,
                            operation, protocol, packetReceived, false, state);
                        LOG.info("onPacketReceived : First Packet IN Queue Size : {}",
                                ((ThreadPoolExecutor)firstPacketExecutorService).getQueue().size());
                        firstPacketExecutorService.execute(() -> naptEventHandler.handleEvent(naptEntryEvent));
                    } else {
                        LOG.trace("onPacketReceived : SNAT({}) Packet already processed.", sourceIPPortKey);
                        NAPTEntryEvent naptEntryEvent = new NAPTEntryEvent(internalIPAddress, portNumber, routerId,
                            operation, protocol, packetReceived, true, state);
                        LOG.info("onPacketReceived : Retry Packet IN Queue Size : {}",
                                ((ThreadPoolExecutor)retryPacketExecutorService).getQueue().size());

                        long firstPacketInTime = state.getFirstPacketInTime();
                        retryPacketExecutorService.execute(() -> {
                            if (System.currentTimeMillis() - firstPacketInTime > 4000) {
                                LOG.error("onPacketReceived : Flow not installed even after 4sec."
                                        + "Dropping SNAT ({}) Packet", sourceIPPortKey);
                                removeIncomingPacketMap(sourceIPPortKey);
                                return;
                            }
                            naptEventHandler.handleEvent(naptEntryEvent);<commit_msg>Move sometimes verbose log from INFO to DEBUG<commit_after>/*
 * Copyright (c) 2016 Ericsson India Global Services Pvt Ltd. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */
package org.opendaylight.netvirt.natservice.internal;

import com.google.common.primitives.Ints;
import java.math.BigInteger;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import javax.inject.Singleton;
import org.opendaylight.controller.liblldp.NetUtils;
import org.opendaylight.genius.mdsalutil.MetaDataUtil;
import org.opendaylight.genius.mdsalutil.NWUtil;
import org.opendaylight.genius.mdsalutil.NwConstants;
import org.opendaylight.genius.mdsalutil.packet.Ethernet;
import org.opendaylight.genius.mdsalutil.packet.IPv4;
import org.opendaylight.genius.mdsalutil.packet.TCP;
import org.opendaylight.genius.mdsalutil.packet.UDP;
import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketProcessingListener;
import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Singleton
public class NaptPacketInHandler implements PacketProcessingListener {

    private static final Logger LOG = LoggerFactory.getLogger(NaptPacketInHandler.class);
    private final ConcurrentMap<String,NatPacketProcessingState> incomingPacketMap = new ConcurrentHashMap<>();
    private final NaptEventHandler naptEventHandler;
    private ExecutorService firstPacketExecutorService;
    private ExecutorService retryPacketExecutorService;

    @Inject
    public NaptPacketInHandler(NaptEventHandler naptEventHandler) {
        this.naptEventHandler = naptEventHandler;
    }

    @PostConstruct
    public void init() {
        firstPacketExecutorService = Executors.newFixedThreadPool(NatConstants.SNAT_PACKET_THEADPOOL_SIZE);
        retryPacketExecutorService = Executors.newFixedThreadPool(NatConstants.SNAT_PACKET_RETRY_THEADPOOL_SIZE);
    }

    @PreDestroy
    public void close() {
        if (firstPacketExecutorService != null) {
            firstPacketExecutorService.shutdown();
        }

        if (retryPacketExecutorService != null) {
            retryPacketExecutorService.shutdown();
        }
    }

    @Override
    // TODO Clean up the exception handling
    @SuppressWarnings("checkstyle:IllegalCatch")
    public void onPacketReceived(PacketReceived packetReceived) {
        String internalIPAddress = "";
        int portNumber = 0;
        long routerId = 0L;
        NAPTEntryEvent.Operation operation = NAPTEntryEvent.Operation.ADD;
        NAPTEntryEvent.Protocol protocol;

        Short tableId = packetReceived.getTableId().getValue();

        LOG.trace("onPacketReceived : packet: {}, tableId {}", packetReceived, tableId);

        if (tableId == NwConstants.OUTBOUND_NAPT_TABLE) {
            LOG.debug("onPacketReceived : NAPTPacketInHandler Packet for Outbound NAPT Table");
            byte[] inPayload = packetReceived.getPayload();
            Ethernet ethPkt = new Ethernet();
            if (inPayload != null) {
                try {
                    ethPkt.deserialize(inPayload, 0, inPayload.length * NetUtils.NumBitsInAByte);
                } catch (Exception e) {
                    LOG.warn("onPacketReceived: Failed to decode Packet", e);
                    return;
                }
                if (ethPkt.getPayload() instanceof IPv4) {
                    IPv4 ipPkt = (IPv4) ethPkt.getPayload();
                    byte[] ipSrc = Ints.toByteArray(ipPkt.getSourceAddress());

                    internalIPAddress = NWUtil.toStringIpAddress(ipSrc);
                    LOG.trace("onPacketReceived : Retrieved internalIPAddress {}", internalIPAddress);
                    if (ipPkt.getPayload() instanceof TCP) {
                        TCP tcpPkt = (TCP) ipPkt.getPayload();
                        portNumber = tcpPkt.getSourcePort();
                        if (portNumber < 0) {
                            portNumber = 32767 + portNumber + 32767 + 2;
                            LOG.trace("onPacketReceived : Retrieved and extracted TCP portNumber {}", portNumber);
                        }
                        protocol = NAPTEntryEvent.Protocol.TCP;
                        LOG.trace("onPacketReceived : Retrieved TCP portNumber {}", portNumber);
                    } else if (ipPkt.getPayload() instanceof UDP) {
                        UDP udpPkt = (UDP) ipPkt.getPayload();
                        portNumber = udpPkt.getSourcePort();
                        if (portNumber < 0) {
                            portNumber = 32767 + portNumber + 32767 + 2;
                            LOG.trace("onPacketReceived : Retrieved and extracted UDP portNumber {}", portNumber);
                        }
                        protocol = NAPTEntryEvent.Protocol.UDP;
                        LOG.trace("onPacketReceived : Retrieved UDP portNumber {}", portNumber);
                    } else {
                        LOG.error("onPacketReceived : Incoming Packet is neither TCP or UDP packet");
                        return;
                    }
                } else {
                    LOG.error("onPacketReceived : Incoming Packet is not IPv4 packet");
                    return;
                }

                if (internalIPAddress != null) {
                    BigInteger metadata = packetReceived.getMatch().getMetadata().getMetadata();
                    routerId = MetaDataUtil.getNatRouterIdFromMetadata(metadata);
                    if (routerId <= 0) {
                        LOG.error("onPacketReceived : Router ID is invalid");
                        return;
                    }
                    String sourceIPPortKey = routerId + NatConstants.COLON_SEPARATOR
                            + internalIPAddress + NatConstants.COLON_SEPARATOR + portNumber;

                    NatPacketProcessingState state = incomingPacketMap.get(sourceIPPortKey);
                    if (state == null) {
                        state = new NatPacketProcessingState(System.currentTimeMillis(), -1);
                        incomingPacketMap.put(sourceIPPortKey, state);
                        LOG.trace("onPacketReceived : Processing new SNAT({}) Packet", sourceIPPortKey);

                        //send to Event Queue
                        NAPTEntryEvent naptEntryEvent = new NAPTEntryEvent(internalIPAddress, portNumber, routerId,
                            operation, protocol, packetReceived, false, state);
                        LOG.info("onPacketReceived : First Packet IN Queue Size : {}",
                                ((ThreadPoolExecutor)firstPacketExecutorService).getQueue().size());
                        firstPacketExecutorService.execute(() -> naptEventHandler.handleEvent(naptEntryEvent));
                    } else {
                        LOG.trace("onPacketReceived : SNAT({}) Packet already processed.", sourceIPPortKey);
                        NAPTEntryEvent naptEntryEvent = new NAPTEntryEvent(internalIPAddress, portNumber, routerId,
                            operation, protocol, packetReceived, true, state);
                        LOG.debug("onPacketReceived : Retry Packet IN Queue Size : {}",
                                ((ThreadPoolExecutor)retryPacketExecutorService).getQueue().size());

                        long firstPacketInTime = state.getFirstPacketInTime();
                        retryPacketExecutorService.execute(() -> {
                            if (System.currentTimeMillis() - firstPacketInTime > 4000) {
                                LOG.error("onPacketReceived : Flow not installed even after 4sec."
                                        + "Dropping SNAT ({}) Packet", sourceIPPortKey);
                                removeIncomingPacketMap(sourceIPPortKey);
                                return;
                            }
                            naptEventHandler.handleEvent(naptEntryEvent);
{'fraction_non_alphanumeric': 0.0829919857524488, 'fraction_numerical': 0.007479964381121995, 'mean_word_length': 2.0966911764705882, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2899401', 'n_tokens_mistral': 4327, 'n_tokens_neox': 4107, 'n_words': 860}